From aa892e17fbaf037cecaa05066c1c136a9aab5ca4 Mon Sep 17 00:00:00 2001 From: Ching L Date: Fri, 12 Sep 2025 17:58:02 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8F=91=E5=B8=83=20v4.0=EF=BC=9A=E4=BB=A3?= =?UTF-8?q?=E7=A0=81=E7=BB=93=E6=9E=84=E9=87=8D=E6=9E=84=E4=B8=8E=E9=94=99?= =?UTF-8?q?=E8=AF=AF=E5=A4=84=E7=90=86=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 主要改进： - 使用 dataclass 替代字典存储比赛数据，提高类型安全性 - 实现自动重试机制，添加指数退避 - 使用 Python logging 模块，支持多级别日志 - 拆分大函数为多个专门的方法，提高可维护性 - 实现优雅降级，单个失败不影响整体流程 - 将 v3 版本移至 legacy 目录存档 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude --- CHANGELOG.md | 31 +- README.md | 16 +- legacy/sync_dota2_matches_v3.py | 1241 +++++++++++++++++++++++++++ sync_dota2_matches.py | 1423 +++++++++++++++---------------- 4 files changed, 1966 insertions(+), 745 deletions(-) create mode 100644 legacy/sync_dota2_matches_v3.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 1f4ec10..6ab4090 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,27 @@ # Changelog +## v4.0 - 2025-09-12 - 代码结构重构与错误处理优化 +- **🏗️ 代码结构重构**： + - 使用 `@dataclass` 替代字典存储比赛数据，提高类型安全性 + - 拆分大函数 `sync_matches_to_calendar()` 为多个专门的方法 + - 新增 `process_upcoming_matches()` 处理即将进行的比赛 + - 新增 `process_completed_matches()` 处理已完成的比赛 + - 新增 `clean_duplicate_and_expired_events()` 清理重复和过期事件 + - 提取 `find_existing_event()` 统一事件查找逻辑 +- **🔄 增强错误处理**： + - 实现 `retry_on_exception` 装饰器，提供指数退避重试机制 + - 所有网络请求（Liquipedia 和 Google Calendar API）都支持自动重试 + - 单个比赛处理失败不影响整体同步流程（优雅降级） +- **📝 专业日志系统**： + - 使用 Python `logging` 模块替代 `print` 语句 + - 支持多级别日志（DEBUG, INFO, WARNING, ERROR） + - 添加时间戳和详细的异常堆栈信息 + - 支持通过命令行参数调整日志级别 +- **⚡ 性能和可维护性**： + - 提取重复代码为独立方法（`_extract_score()`, `_extract_format()`, `_extract_tournament()`） + - 改进的错误报告和调试信息 + - 更好的代码组织和模块化设计 + ## v3.9 - 2025-09-12 - 增强TBD事件清理功能 - **自动删除已结束的TBD事件**： - 检查事件的结束时间，如果已过且仍包含TBD则自动删除 @@ -157,15 +179,16 @@ | v3.7 | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ | | v3.8 | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ | | v3.9 | ✓ | ✓ | ✓ | ✓ | ✓+ | ✓ | +| v4.0 | ✓ | ✓ | ✓ | ✓ | ✓+ | ✓ | ## 使用建议 -推荐使用最新的 v3 版本，它包含所有功能： +推荐使用最新的 v4.0 版本，它包含所有功能并提供更好的错误处理： ```bash -./run_sync.sh +python sync_dota2_matches.py --calendar-id "YOUR_CALENDAR_ID" ``` -或手动运行： +调试模式： ```bash -python sync_dota2_matches_v3.py --calendar-id "YOUR_CALENDAR_ID" +python sync_dota2_matches.py --calendar-id "YOUR_CALENDAR_ID" --log-level DEBUG ``` \ No newline at end of file diff --git a/README.md b/README.md index 48aef42..e9b515d 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,14 @@ -# Dota 2 Calendar Sync v3.9 +# Dota 2 Calendar Sync v4.0 自动从 Liquipedia 获取 Dota 2 Tier 1 比赛信息并同步到 Google Calendar，支持自动更新比赛结果、时间变更、智能管理TBD占位事件、自动清理过期和重复比赛。 +**v4.0 新特性**： +- 🏗️ 代码结构重构：使用 dataclass 替代字典，提高类型安全性 +- 🔄 增强错误处理：添加自动重试机制（指数退避） +- 📝 专业日志系统：使用 Python logging 模块，支持多级别日志 +- 🛡️ 优雅降级：部分失败不影响整体同步流程 +- ⚡ 代码模块化：拆分大函数，提高可维护性 + ## 功能 - 自动获取 Liquipedia 上的 Tier 1 级别 Dota 2 比赛 @@ -49,6 +56,9 @@ python sync_dota2_matches.py # 指定特定的 Google Calendar python sync_dota2_matches.py --calendar-id "091325d4ea74ad78387402db1a428390c4779dff573322863b6fca00194da024@group.calendar.google.com" + +# 使用调试模式查看详细日志 +python sync_dota2_matches.py --log-level DEBUG ``` ### Dry Run 模式 @@ -66,10 +76,12 @@ python sync_dota2_matches.py --dry-run - `--no-results`: 跳过更新已完成比赛的结果 - `--no-time-updates`: 跳过更新比赛时间变更 - `--credentials`: 服务账号凭据文件路径（默认: credentials.json） +- `--log-level`: 设置日志级别 (DEBUG, INFO, WARNING, ERROR) ## 文件说明 -- `sync_dota2_matches.py`: 主同步脚本（v3版本） +- `sync_dota2_matches.py`: 主同步脚本（v4版本 - 优化版） +- `legacy/`: 历史版本脚本存档 - `credentials.json`: Google 服务账号凭据（需要自行添加） - `requirements.txt`: Python 依赖包 - `run_sync.sh`: 便捷运行脚本 diff --git a/legacy/sync_dota2_matches_v3.py b/legacy/sync_dota2_matches_v3.py new file mode 100644 index 0000000..8c0996f --- /dev/null +++ b/legacy/sync_dota2_matches_v3.py @@ -0,0 +1,1241 @@ +#!/usr/bin/env python3 +""" +Dota 2 Tournament Calendar Sync v3 +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 +""" + +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 + +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) + print(f"✓ Successfully authenticated with Google Calendar") + return service + except Exception as e: + print(f"✗ Authentication failed: {e}") + sys.exit(1) + + def fetch_all_matches(self): + """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)' + } + + print(f"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.get('completed', False)] + completed = [m for m in matches if m.get('completed', False)] + + print(f"✓ Found {len(upcoming)} upcoming matches") + print(f"✓ Found {len(completed)} completed matches with results") + + return upcoming, completed + + except requests.RequestException as e: + print(f"✗ Error fetching Liquipedia data: {e}") + return [], [] + + def _parse_match(self, parent, timestamp_elem): + """Parse match data from an element using HTML structure""" + try: + match_data = {} + + # Get timestamp + timestamp = timestamp_elem.get('data-timestamp') + if timestamp: + match_data['datetime'] = datetime.fromtimestamp(int(timestamp), tz=pytz.UTC) + else: + return None + + # Extract team names from HTML structure + team_blocks = parent.find_all('div', class_='block-team') + if len(team_blocks) >= 2: + # Get team names - prefer span.name over a tag (a tag might be empty icon link) + 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: + match_data['team1'] = self._clean_team_name(team1_elem.get_text().strip()) + match_data['team2'] = self._clean_team_name(team2_elem.get_text().strip()) + + # If team blocks not found, try fallback + if 'team1' not in match_data: + # Fallback to text parsing + text = parent.get_text() + # Look for "vs" pattern + 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() + # Clean up team names + team1_raw = re.sub(r'^.*CEST?', '', team1_raw).strip() + match_data['team1'] = self._clean_team_name(team1_raw) + match_data['team2'] = self._clean_team_name(team2_raw) + + # Extract score from HTML structure + has_score = False + score_match = None + + # 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: + has_score = True + match_data['score'] = f"{score1}-{score2}" + score_match = True # Use as flag + except ValueError: + pass + + # If score not found in structure, try text pattern + if not has_score: + text = parent.get_text() + # Only look for dash pattern (not colon) to avoid matching time + 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)) + # Validate it's a reasonable game score + if 0 <= score1 <= 5 and 0 <= score2 <= 5 and (score1 + score2) > 0: + # Additional check: make sure this isn't part of a date (e.g., 2025-01-14) + 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): + has_score = True + match_data['score'] = f"{score1}-{score2}" + score_match = score_pattern + + # Extract format (Bo1, Bo3, Bo5) + 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: + match_data['format'] = format_match.group(1) + else: + # Fallback to text search + text = parent.get_text() + format_match = re.search(r'\(?(Bo\d)\)?', text) + if format_match: + match_data['format'] = format_match.group(1) + + # Extract tournament from HTML structure + tournament_elem = parent.find('div', class_='match-info-tournament') + if tournament_elem: + tournament_text = tournament_elem.get_text().strip() + # Clean up tournament name + tournament_text = re.sub(r'\+ Add details.*', '', tournament_text).strip() + if 'TI2025' in tournament_text: + match_data['tournament'] = 'The International 2025' + round_match = re.search(r'Round\s+\d+', tournament_text) + if round_match: + match_data['tournament'] += f" - {round_match.group(0)}" + else: + match_data['tournament'] = tournament_text + else: + # Fallback to text search + text = parent.get_text() + if 'TI2025' in text: + match_data['tournament'] = 'The International 2025' + round_match = re.search(r'Round\s+\d+', text) + if round_match: + match_data['tournament'] += f" - {round_match.group(0)}" + elif 'Major' in text: + major_match = re.search(r'[\w\s]+Major', text) + if major_match: + match_data['tournament'] = major_match.group(0).strip() + + # Mark if has score and if completed + if has_score: + # TBD vs TBD matches should NEVER be marked as having a score or completed + if match_data.get('team1') == 'TBD' and match_data.get('team2') == 'TBD': + has_score = False + match_data['completed'] = False + match_data['has_score'] = False + # Remove any incorrectly parsed score + if 'score' in match_data: + del match_data['score'] + else: + # Score already set above, extract score values + score_parts = re.match(r'(\d+)-(\d+)', match_data['score']) + if score_parts: + score1 = int(score_parts.group(1)) + score2 = int(score_parts.group(2)) + else: + score1 = score2 = 0 + + # Check if series is actually completed based on format + series_completed = False + if 'format' in match_data: + if 'Bo3' in match_data['format']: + # Bo3 is complete when someone reaches 2 wins + series_completed = (score1 >= 2 or score2 >= 2) + elif 'Bo5' in match_data['format']: + # Bo5 is complete when someone reaches 3 wins + series_completed = (score1 >= 3 or score2 >= 3) + elif 'Bo1' in match_data['format']: + # Bo1 is complete when there's any score + series_completed = True + else: + # Unknown format, assume completed if there's a score + series_completed = True + else: + # No format info, try to guess from score + # If someone has 2+ wins, likely a completed Bo3/Bo5 + series_completed = (score1 >= 2 or score2 >= 2) + + match_data['completed'] = series_completed + match_data['has_score'] = True # Mark that there's a score even if not completed + + # Determine winner only if completed + if series_completed: + if score1 > score2: + match_data['winner'] = match_data.get('team1', 'Unknown') + else: + match_data['winner'] = match_data.get('team2', 'Unknown') + else: + match_data['completed'] = False + match_data['has_score'] = False + + # Generate ID if we have valid data + if 'team1' in match_data and 'team2' in match_data: + match_data['id'] = self._generate_match_id(match_data) + return match_data + + except Exception as e: + pass + + return None + + def _clean_team_name(self, name): + """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, match_data): + """Generate a unique ID for a match""" + # Use teams and tournament for ID (not datetime to handle reschedules) + id_parts = [] + + # For TBD vs TBD matches, include datetime to make them unique + if match_data.get('team1') == 'TBD' and match_data.get('team2') == 'TBD': + # Include datetime for TBD matches to avoid duplicates + if 'datetime' in match_data: + id_parts.append(str(match_data['datetime'])) + if 'tournament' in match_data: + id_parts.append(match_data['tournament']) + else: + # Normal matches: use teams and tournament + if 'team1' in match_data: + id_parts.append(match_data['team1']) + if 'team2' in match_data: + id_parts.append(match_data['team2']) + if 'tournament' in match_data: + id_parts.append(match_data['tournament']) + else: + # Fall back to date if no tournament + if 'datetime' in match_data: + id_parts.append(str(match_data['datetime'].date())) + + unique_string = '_'.join(id_parts) + return hashlib.md5(unique_string.encode()).hexdigest()[:16] + + def _remove_duplicates(self, matches): + """Remove duplicate matches based on ID""" + unique_matches = {} + for match in matches: + if match.get('id'): + unique_matches[match['id']] = match + return list(unique_matches.values()) + + def get_existing_events(self, days_back=7, days_ahead=30): + """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' + + print(f"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 + # Build multiple indexes for better matching + dota_events_by_id = {} + dota_events_by_match = {} + + for event in events: + summary = event.get('summary', '') + # Check for Dota events - old format has "Dota 2", new format has tournament brackets + is_dota = ('Dota 2' in summary or + 'The International' in summary or + 'TI2025' in summary or + '[' in summary and 'vs' in summary) # New format has brackets + + if is_dota: + description = event.get('description', '') + + # Extract ID from description (for old events) + id_match = re.search(r'ID:\s*([a-f0-9]+)', description) + if id_match: + dota_events_by_id[id_match.group(1)] = event + + # Also create key based on teams and tournament for matching + summary = event.get('summary', '') + # Remove completed markers and scores + summary = summary.replace('[COMPLETED] ', '') + # Remove checkmark and score (format: "✓ 2-1 Team vs Team") + summary = re.sub(r'^✓\s+\d+[-:]\d+\s+', '', summary) + # Also handle old format with score at end + summary = re.sub(r'\s*\([0-9\-\?]+\)\s*$', '', summary) + + # Try new format first: "Team1 vs Team2 [Tournament]" + match = re.search(r'^(.*?)\s+vs\s+(.*?)\s*\[(.*?)\]$', summary) + if match: + team1 = match.group(1).strip() + team2 = match.group(2).strip() + tournament = match.group(3).strip() + else: + # Try old format: "Dota 2 - Tournament: Team1 vs Team2" + 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() + + if match: + + # Create match key (teams + tournament) + match_key = f"{team1}_{team2}_{tournament}" + dota_events_by_match[match_key] = event + + print(f"✓ Found {len(dota_events_by_id)} existing Dota 2 events") + + # Return combined dictionary for backward compatibility + combined = {} + combined.update(dota_events_by_id) + # Store the by_match index as a special key + combined['_by_match'] = dota_events_by_match + + return combined + + except Exception as e: + print(f"✗ Error fetching calendar events: {e}") + return {} + + def create_calendar_event(self, match_data): + """Create a Google Calendar event for a match""" + team1 = match_data.get('team1', 'TBD') + team2 = match_data.get('team2', 'TBD') + tournament = match_data.get('tournament', '') + + # New format: Teams first, tournament in brackets + if tournament: + summary = f"{team1} vs {team2} [{tournament}]" + else: + summary = f"{team1} vs {team2}" + + # Build description + description_parts = [] + if tournament: + description_parts.append(f"Tournament: {tournament}") + description_parts.append(f"Match: {team1} vs {team2}") + if 'format' in match_data: + description_parts.append(f"Format: {match_data['format']}") + if match_data.get('completed'): + description_parts.append(f"\n🏆 RESULT: {match_data.get('score', 'Unknown')}") + description_parts.append(f"Winner: {match_data.get('winner', 'Unknown')}") + description_parts.append(f"ID: {match_data['id']}") + description_parts.append("\nSource: Liquipedia") + + description = '\n'.join(description_parts) + + # Set start and end times + start_time = match_data.get('datetime', datetime.now(pytz.UTC)) + # Estimate duration + duration = 2 + if 'format' in match_data: + if 'Bo5' in match_data['format']: + duration = 4 + elif 'Bo3' in match_data['format']: + duration = 3 + elif 'Bo1' in match_data['format']: + duration = 1 + + end_time = start_time + timedelta(hours=duration) + + event = { + 'summary': summary, + 'description': description, + 'start': { + 'dateTime': start_time.isoformat(), + 'timeZone': 'UTC', + }, + 'end': { + 'dateTime': end_time.isoformat(), + 'timeZone': 'UTC', + }, + 'reminders': { + 'useDefault': False, + 'overrides': [ + {'method': 'popup', 'minutes': 30}, + ], + }, + 'colorId': '9', # Blue + } + + return event + + def update_event_time(self, event_id, new_datetime): + """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 to description + description = event.get('description', '') + if 'Last updated:' in description: + # Update the timestamp + description = re.sub( + r'Last updated:.*', + f"Last updated: {datetime.now(pytz.UTC).strftime('%Y-%m-%d %H:%M UTC')}", + description + ) + else: + # Add timestamp + description += f"\nLast updated: {datetime.now(pytz.UTC).strftime('%Y-%m-%d %H:%M UTC')}" + event['description'] = description + + # Update the event + updated_event = self.service.events().update( + calendarId=self.calendar_id, + eventId=event_id, + body=event + ).execute() + + return True + + except Exception as e: + print(f"Error updating event time: {e}") + return False + + def update_event_with_teams(self, event_id, match_data): + """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() + + team1 = match_data.get('team1', 'TBD') + team2 = match_data.get('team2', 'TBD') + tournament = match_data.get('tournament', '') + + # Update summary with actual team names + if tournament: + new_summary = f"{team1} vs {team2} [{tournament}]" + else: + new_summary = f"{team1} vs {team2}" + + # Update description + description = event.get('description', '') + # Update the Match line + description = re.sub( + r'Match: .*?\n', + f"Match: {team1} vs {team2}\n", + description + ) + # Update the ID to the new one + if 'ID:' in description: + description = re.sub( + r'ID: [a-f0-9]+', + f"ID: {match_data.get('id', '')}", + description + ) + + # Add update timestamp + if 'Teams updated:' in description: + description = re.sub( + r'Teams updated:.*', + f"Teams updated: {datetime.now(pytz.UTC).strftime('%Y-%m-%d %H:%M UTC')}", + description + ) + else: + description = description.replace('\nSource:', f"\nTeams updated: {datetime.now(pytz.UTC).strftime('%Y-%m-%d %H:%M UTC')}\nSource:") + + event['summary'] = new_summary + event['description'] = description + + # Update the event + updated_event = self.service.events().update( + calendarId=self.calendar_id, + eventId=event_id, + body=event + ).execute() + + return True + + except Exception as e: + print(f"Error updating event with teams: {e}") + return False + + def update_event_with_score(self, event_id, match_data): + """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 the description with current score + description = event.get('description', '') + + # Check if score is already in the description + if '📊 CURRENT SCORE:' in description: + # Update existing score + description = re.sub( + r'📊 CURRENT SCORE:.*?\n', + f"📊 CURRENT SCORE: {match_data.get('score', 'Unknown')}\n", + description + ) + else: + # Add new score + score_text = f"\n📊 CURRENT SCORE: {match_data.get('score', 'Unknown')}\n" + if 'ID:' in description: + description = description.replace('ID:', score_text + 'ID:') + else: + description += score_text + + # Update the summary to show current score (without checkmark) + summary = event.get('summary', '') + # Remove any existing score (including multiple scores) + summary = re.sub(r'^(\d+[-:]\d+\s+)+', '', summary) + # Add new score at the beginning + score = match_data.get('score', '?-?') + summary = f"{score} {summary}" + + # Update the event + event['description'] = description + event['summary'] = summary + + updated_event = self.service.events().update( + calendarId=self.calendar_id, + eventId=event_id, + body=event + ).execute() + + return True + + except Exception as e: + print(f"Error updating event score: {e}") + return False + + def update_event_with_result(self, event_id, match_data): + """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 the description with results + description = event.get('description', '') + + # Check if results are already in the description + if '🏆 RESULT:' in description: + # Update existing result + description = re.sub( + r'🏆 RESULT:.*?\n.*?Winner:.*?\n', + f"🏆 RESULT: {match_data.get('score', 'Unknown')}\nWinner: {match_data.get('winner', 'Unknown')}\n", + description, + flags=re.DOTALL + ) + else: + # Add new result + result_text = f"\n🏆 RESULT: {match_data.get('score', 'Unknown')}\nWinner: {match_data.get('winner', 'Unknown')}\n" + if 'ID:' in description: + description = description.replace('ID:', result_text + 'ID:') + else: + description += result_text + + # Update the summary to show it's completed with result + summary = event.get('summary', '') + # First remove any existing scores (in-progress scores) + summary = re.sub(r'^(\d+[-:]\d+\s+)+', '', summary) + # Remove any existing checkmark + summary = re.sub(r'^✓\s+', '', summary) + # Add checkmark and final score + score = match_data.get('score', '?-?') + summary = f"✓ {score} {summary}" + + # Update the event + event['description'] = description + event['summary'] = summary + + updated_event = self.service.events().update( + calendarId=self.calendar_id, + eventId=event_id, + body=event + ).execute() + + return True + + except Exception as e: + print(f"Error updating event: {e}") + return False + + def check_time_difference(self, event_datetime, new_datetime): + """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 # Return True if difference is 5 minutes or more + + def delete_calendar_event(self, event_id): + """Delete a calendar event""" + try: + self.service.events().delete( + calendarId=self.calendar_id, + eventId=event_id + ).execute() + return True + except Exception as e: + print(f"Error deleting event: {e}") + return False + + def sync_matches_to_calendar(self, dry_run=False, update_results=True, update_times=True, delete_old_tbd=True): + """Main sync function with time updates""" + print("\n" + "="*50) + print("Starting Dota 2 Calendar Sync v3") + print("="*50 + "\n") + + # Fetch all matches + upcoming_matches, completed_matches = self.fetch_all_matches() + + if not upcoming_matches and not completed_matches: + print("No matches found to sync") + return + + # Get existing events + existing_events = self.get_existing_events(days_back=7, days_ahead=30) + + # Counters + added_count = 0 + skipped_count = 0 + updated_count = 0 + time_updated_count = 0 + error_count = 0 + deleted_tbd_count = 0 + + # Track which TBD events were updated + updated_tbd_events = set() + + # Process upcoming matches + print("\nProcessing upcoming matches...") + print("-" * 30) + + now = datetime.now(pytz.UTC) + # Include matches from the last 12 hours (to catch ongoing matches with scores) + twelve_hours_ago = now - timedelta(hours=12) + future_matches = [m for m in upcoming_matches + if m.get('datetime', now) >= twelve_hours_ago] + + for match in future_matches: + match_id = match.get('id') + team1 = match.get('team1', 'TBD') + team2 = match.get('team2', 'TBD') + match_time = match.get('datetime', now) + tournament = match.get('tournament', '') + + if not match_id: + continue + + # Try to find existing event by ID or by match details + existing_event = None + + # First try by ID + if match_id in existing_events: + existing_event = existing_events[match_id] + + # If not found, try by match details (teams + tournament) + if not existing_event and '_by_match' in existing_events: + match_key = f"{team1}_{team2}_{tournament}" + if match_key in existing_events['_by_match']: + existing_event = existing_events['_by_match'][match_key] + + # Also try to find by teams only (ignoring score) for live updates + # This handles cases where score changes during match + if not existing_event: + for event_key, event in existing_events['_by_match'].items(): + # Check if teams match (order independent) + if (f"{team1}_{team2}" in event_key or f"{team2}_{team1}" in event_key) and tournament in event_key: + existing_event = event + print(f" → Found existing match by teams: {team1} vs {team2}") + break + + # Special handling for TBD matches that might have been updated + # Look for TBD events at the same time and tournament + if not existing_event and '_by_match' in existing_events: + # Only look for TBD to update if current match is NOT TBD vs TBD + # (we don't want to match TBD vs TBD with other TBD vs TBD) + if not (team1 == 'TBD' and team2 == 'TBD'): + # Check if this match used to be TBD + for event_key, event in existing_events['_by_match'].items(): + if 'TBD_TBD' in event_key and 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')) + # Relaxed time matching: within 1 hour (3600 seconds) + if abs((event_dt - match_time).total_seconds()) < 3600: # Within 1 hour + existing_event = event + print(f" → Found TBD match to update: {team1} vs {team2}") + print(f" Time difference: {abs((event_dt - match_time).total_seconds())/60:.0f} minutes") + break + + 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 (team1 != 'TBD' or team2 != 'TBD') + + if is_tbd_update: + # Update TBD match with actual teams + if dry_run: + print(f"◯ Would update TBD match with teams: {team1} vs {team2}") + updated_count += 1 + else: + if self.update_event_with_teams(existing_event['id'], match): + print(f"✓ Updated TBD match with teams: {team1} vs {team2}") + updated_count += 1 + updated_tbd_events.add(existing_event['id']) + time.sleep(0.2) + else: + print(f"✗ Failed to update TBD match: {team1} vs {team2}") + error_count += 1 + # Check if this match has a score (completed or in-progress) and needs update + elif match.get('has_score') and update_results: + # Check current event status + summary = existing_event.get('summary', '') + description = existing_event.get('description', '') + current_score = None + + # Try to extract current score from summary + score_in_summary = re.search(r'✓?\s*(\d+[-:]\d+)', summary) + if score_in_summary: + current_score = score_in_summary.group(1).replace(':', '-') + + # Check if score needs update + new_score = match.get('score', 'Unknown') + + if current_score == new_score: + print(f"⊘ Score unchanged: {team1} vs {team2} ({new_score})") + skipped_count += 1 + else: + if match.get('completed'): + # Series is completed + if dry_run: + print(f"◯ Would update completed result: {team1} vs {team2} - {new_score}") + updated_count += 1 + else: + if self.update_event_with_result(existing_event['id'], match): + print(f"✓ Updated completed result: {team1} vs {team2} - {new_score}") + updated_count += 1 + time.sleep(0.2) + else: + print(f"✗ Failed to update: {team1} vs {team2}") + error_count += 1 + else: + # Series is in-progress with partial score + if dry_run: + print(f"◯ Would update in-progress score: {team1} vs {team2} - {new_score}") + updated_count += 1 + else: + if self.update_event_with_score(existing_event['id'], match): + print(f"📊 Updated in-progress score: {team1} vs {team2} - {new_score}") + updated_count += 1 + time.sleep(0.2) + else: + print(f"✗ Failed to update score: {team1} vs {team2}") + error_count += 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_time): + # Time has changed + if dry_run: + old_time = datetime.fromisoformat(event_start.replace('Z', '+00:00')) + print(f"◯ Would update time: {team1} vs {team2}") + print(f" Old: {old_time.strftime('%Y-%m-%d %H:%M UTC')}") + print(f" New: {match_time.strftime('%Y-%m-%d %H:%M UTC')}") + time_updated_count += 1 + else: + old_time = datetime.fromisoformat(event_start.replace('Z', '+00:00')) + if self.update_event_time(existing_event['id'], match_time): + print(f"⏰ Updated time: {team1} vs {team2}") + print(f" Old: {old_time.strftime('%Y-%m-%d %H:%M UTC')}") + print(f" New: {match_time.strftime('%Y-%m-%d %H:%M UTC')}") + time_updated_count += 1 + time.sleep(0.2) + else: + print(f"✗ Failed to update time: {team1} vs {team2}") + error_count += 1 + else: + print(f"⊘ No change: {team1} vs {team2}") + skipped_count += 1 + else: + print(f"⊘ Skipping (exists): {team1} vs {team2}") + skipped_count += 1 + else: + # New match + if dry_run: + print(f"◯ Would add: {team1} vs {team2} at {match_time.strftime('%Y-%m-%d %H:%M UTC')}") + added_count += 1 + else: + try: + event = self.create_calendar_event(match) + self.service.events().insert( + calendarId=self.calendar_id, + body=event + ).execute() + print(f"✓ Added: {team1} vs {team2} at {match_time.strftime('%Y-%m-%d %H:%M UTC')}") + added_count += 1 + time.sleep(0.2) + except Exception as e: + print(f"✗ Error adding {team1} vs {team2}: {e}") + error_count += 1 + + # Process completed matches for results + if update_results and completed_matches: + print("\nProcessing completed match results...") + print("-" * 30) + + for match in completed_matches: + match_id = match.get('id') + team1 = match.get('team1', 'TBD') + team2 = match.get('team2', 'TBD') + score = match.get('score', 'Unknown') + tournament = match.get('tournament', '') + + if not match_id: + continue + + # Try to find existing event by ID or by match details + existing_event = None + + # First try by ID + if match_id in existing_events: + existing_event = existing_events[match_id] + + # If not found, try by match details + if not existing_event and '_by_match' in existing_events: + match_key = f"{team1}_{team2}_{tournament}" + if match_key in existing_events['_by_match']: + existing_event = existing_events['_by_match'][match_key] + + # Also try to find by teams only (for live score updates) + if not existing_event: + for event_key, event in existing_events['_by_match'].items(): + # Check if teams match (order independent) and tournament matches + if (f"{team1}_{team2}" in event_key or f"{team2}_{team1}" in event_key) and tournament in event_key: + existing_event = event + print(f" → Found existing match by teams: {team1} vs {team2}") + break + + if existing_event: + # Check if already marked as completed + summary = existing_event.get('summary', '') + if '✓' in summary or '[COMPLETED]' in summary: + print(f"⊘ Already updated: {team1} vs {team2} ({score})") + else: + if dry_run: + print(f"◯ Would update result: {team1} vs {team2} - {score}") + updated_count += 1 + else: + if self.update_event_with_result(existing_event['id'], match): + print(f"✓ Updated result: {team1} vs {team2} - {score}") + updated_count += 1 + time.sleep(0.2) + else: + print(f"✗ Failed to update: {team1} vs {team2}") + error_count += 1 + + # Delete old TBD events that are past and not updated + # Also check for duplicate TBD events at the same time + # Also delete TBD events when a confirmed match exists at the same time + if delete_old_tbd and not dry_run: + print("\nChecking for expired, duplicate, and superseded TBD events to delete...") + print("-" * 30) + + # Group all events by time to find duplicates and superseded TBD events + events_by_time = {} + tbd_by_time = {} + + # Get all events and group 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 for checking if match has ended + 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: + # If no end time, assume 3 hours duration + 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 (end time has passed) and still contains TBD + if event_end_dt < now: + if self.delete_calendar_event(event_id): + print(f"🗑️ Deleted ended TBD event: {summary} (ended at {event_end_dt.strftime('%Y-%m-%d %H:%M UTC')})") + deleted_tbd_count += 1 + time.sleep(0.2) + else: + print(f"✗ Failed to delete ended TBD event: {summary}") + error_count += 1 + continue # Don't process this event further + + # 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 that have been superseded by confirmed matches + for time_key, events in events_by_time.items(): + if events['confirmed'] and events['tbd']: + # We have both confirmed and TBD events at the same time + for tbd_event in events['tbd']: + tbd_summary = tbd_event.get('summary', '') + + # Check if this is a complete TBD vs TBD event + if 'TBD vs TBD' in tbd_summary: + # Delete TBD vs TBD when there's any confirmed match at the same time + # Since TBD vs TBD is a placeholder, any confirmed match supersedes it + if events['confirmed']: + confirmed_event = events['confirmed'][0] # Use first confirmed match for logging + confirmed_summary = confirmed_event.get('summary', '') + if self.delete_calendar_event(tbd_event['id']): + print(f"🗑️ Deleted TBD vs TBD event at same time as confirmed match") + print(f" TBD event: {tbd_summary}") + print(f" Confirmed match: {confirmed_summary}") + deleted_tbd_count += 1 + time.sleep(0.2) + else: + print(f"✗ Failed to delete TBD vs TBD event: {tbd_summary}") + error_count += 1 + else: + # For partial TBD events (one team is known), check for team match + team_match = re.search(r'(\w+)\s+vs\s+TBD|TBD\s+vs\s+(\w+)', tbd_summary) + if team_match: + team_in_tbd = team_match.group(1) or team_match.group(2) + + # Check if this team has a confirmed match + for confirmed_event in events['confirmed']: + confirmed_summary = confirmed_event.get('summary', '') + if team_in_tbd and team_in_tbd in confirmed_summary: + # This TBD event has been superseded + if self.delete_calendar_event(tbd_event['id']): + print(f"🗑️ Deleted superseded TBD event: {tbd_summary}") + print(f" Replaced by: {confirmed_summary}") + deleted_tbd_count += 1 + time.sleep(0.2) + else: + print(f"✗ Failed to delete TBD event: {tbd_summary}") + error_count += 1 + break + + # Delete duplicate TBD vs TBD events at the same time + for time_key, events in tbd_by_time.items(): + if len(events) > 1: + print(f"Found {len(events)} duplicate TBD events at {time_key}") + # Keep the first one, delete the rest + for event in events[1:]: + if self.delete_calendar_event(event['id']): + print(f"🗑️ Deleted duplicate TBD event: {event['summary']}") + deleted_tbd_count += 1 + time.sleep(0.2) + else: + print(f"✗ Failed to delete duplicate TBD event: {event['summary']}") + error_count += 1 + + # Also check for duplicate matches with different completion states + # Group matches by teams and date (not exact time) + matches_by_teams_date = {} + for key, event in existing_events.items(): + if key == '_by_match': + continue + + summary = event.get('summary', '') + # Skip TBD matches + if 'TBD' in summary: + continue + + # Extract teams from summary + # Remove completion markers and scores + clean_summary = re.sub(r'^✓\s*\d+[-:]\d+\s*', '', summary) + clean_summary = re.sub(r'^\d+[-:]\d+\s*', '', clean_summary) + + # Extract teams + teams_match = re.search(r'([\w\s]+)\s+vs\s+([\w\s]+)\s*\[', clean_summary) + if teams_match: + team1 = teams_match.group(1).strip() + team2 = teams_match.group(2).strip() + + # Get date + event_start = event['start'].get('dateTime', event['start'].get('date')) + event_dt = datetime.fromisoformat(event_start.replace('Z', '+00:00')) + date_key = event_dt.strftime('%Y-%m-%d') + + # Create key for this match + match_key = f"{min(team1, team2)}_vs_{max(team1, team2)}_{date_key}" + + if match_key not in matches_by_teams_date: + matches_by_teams_date[match_key] = [] + matches_by_teams_date[match_key].append(event) + + # Delete duplicates, keeping the completed one + for match_key, events in matches_by_teams_date.items(): + if len(events) > 1: + # Sort by completion status (completed first) and time + def sort_key(e): + summary = e.get('summary', '') + is_completed = '✓' in summary + event_start = e['start'].get('dateTime', e['start'].get('date')) + return (not is_completed, event_start) # Completed first, then by time + + sorted_events = sorted(events, key=sort_key) + + # Keep the first (preferably completed) event + print(f"Found {len(events)} duplicate matches: {sorted_events[0]['summary']}") + for event in sorted_events[1:]: + if self.delete_calendar_event(event['id']): + print(f"🗑️ Deleted duplicate match: {event['summary']}") + deleted_tbd_count += 1 + time.sleep(0.2) + else: + print(f"✗ Failed to delete duplicate: {event['summary']}") + error_count += 1 + + # Summary + print("\n" + "="*50) + print("Sync Summary") + print("="*50) + print(f"✓ Added: {added_count} matches") + if time_updated_count > 0: + print(f"⏰ Time updated: {time_updated_count} matches") + if updated_count > 0: + print(f"✓ Results updated: {updated_count} matches") + if deleted_tbd_count > 0: + print(f"🗑️ Deleted: {deleted_tbd_count} expired TBD events") + print(f"⊘ Skipped: {skipped_count} matches (no changes)") + if error_count > 0: + print(f"✗ Errors: {error_count} matches") + + if dry_run: + print("\n⚠ DRY RUN - No actual changes were made") + + print("\n✓ Sync complete!") + +def main(): + parser = argparse.ArgumentParser( + description='Sync Dota 2 Tier 1 matches from Liquipedia to Google Calendar v3' + ) + 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' + ) + + args = parser.parse_args() + + # Notice + print("\n" + "!"*60) + print("Dota 2 Calendar Sync v3") + print("Features: Match sync, result updates, time change detection") + print("Service Account: calendar-bot@tunpok.iam.gserviceaccount.com") + print("!"*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: + print("\n\nSync cancelled by user") + sys.exit(0) + except Exception as e: + print(f"\n✗ Fatal error: {e}") + sys.exit(1) + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/sync_dota2_matches.py b/sync_dota2_matches.py index 8c0996f..a93b305 100644 --- a/sync_dota2_matches.py +++ b/sync_dota2_matches.py @@ -1,11 +1,15 @@ #!/usr/bin/env python3 """ -Dota 2 Tournament Calendar Sync v3 +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 @@ -19,6 +23,82 @@ 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'): @@ -34,20 +114,21 @@ class Dota2CalendarSync: scopes=['https://www.googleapis.com/auth/calendar'] ) service = build('calendar', 'v3', credentials=credentials) - print(f"✓ Successfully authenticated with Google Calendar") + logger.info("✓ Successfully authenticated with Google Calendar") return service except Exception as e: - print(f"✗ Authentication failed: {e}") + logger.error(f"✗ Authentication failed: {e}") sys.exit(1) - def fetch_all_matches(self): + @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)' } - print(f"Fetching matches from Liquipedia...") + logger.info("Fetching matches from Liquipedia...") try: response = requests.get(url, headers=headers, timeout=30) @@ -81,34 +162,35 @@ class Dota2CalendarSync: matches = self._remove_duplicates(matches) # Separate upcoming and completed matches - upcoming = [m for m in matches if not m.get('completed', False)] - completed = [m for m in matches if m.get('completed', False)] + upcoming = [m for m in matches if not m.completed] + completed = [m for m in matches if m.completed] - print(f"✓ Found {len(upcoming)} upcoming matches") - print(f"✓ Found {len(completed)} completed matches with results") + 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: - print(f"✗ Error fetching Liquipedia data: {e}") + logger.error(f"✗ Error fetching Liquipedia data: {e}") return [], [] - def _parse_match(self, parent, timestamp_elem): + def _parse_match(self, parent, timestamp_elem) -> Optional[Match]: """Parse match data from an element using HTML structure""" try: - match_data = {} - # Get timestamp timestamp = timestamp_elem.get('data-timestamp') - if timestamp: - match_data['datetime'] = datetime.fromtimestamp(int(timestamp), tz=pytz.UTC) - else: + 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 (a tag might be empty icon link) + # 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 @@ -126,162 +208,153 @@ class Dota2CalendarSync: break if team1_elem and team2_elem: - match_data['team1'] = self._clean_team_name(team1_elem.get_text().strip()) - match_data['team2'] = self._clean_team_name(team2_elem.get_text().strip()) + 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' not in match_data: + if team1 == 'TBD' and team2 == 'TBD': # Fallback to text parsing text = parent.get_text() - # Look for "vs" pattern 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() - # Clean up team names team1_raw = re.sub(r'^.*CEST?', '', team1_raw).strip() - match_data['team1'] = self._clean_team_name(team1_raw) - match_data['team2'] = self._clean_team_name(team2_raw) + team1 = self._clean_team_name(team1_raw) + team2 = self._clean_team_name(team2_raw) - # Extract score from HTML structure + # 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 - score_match = None + winner = None - # 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: - has_score = True - match_data['score'] = f"{score1}-{score2}" - score_match = True # Use as flag - except ValueError: - pass - - # If score not found in structure, try text pattern - if not has_score: - text = parent.get_text() - # Only look for dash pattern (not colon) to avoid matching time - 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)) - # Validate it's a reasonable game score - if 0 <= score1 <= 5 and 0 <= score2 <= 5 and (score1 + score2) > 0: - # Additional check: make sure this isn't part of a date (e.g., 2025-01-14) - 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): - has_score = True - match_data['score'] = f"{score1}-{score2}" - score_match = score_pattern - - # Extract format (Bo1, Bo3, Bo5) - 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: - match_data['format'] = format_match.group(1) - else: - # Fallback to text search - text = parent.get_text() - format_match = re.search(r'\(?(Bo\d)\)?', text) - if format_match: - match_data['format'] = format_match.group(1) - - # Extract tournament from HTML structure - tournament_elem = parent.find('div', class_='match-info-tournament') - if tournament_elem: - tournament_text = tournament_elem.get_text().strip() - # Clean up tournament name - tournament_text = re.sub(r'\+ Add details.*', '', tournament_text).strip() - if 'TI2025' in tournament_text: - match_data['tournament'] = 'The International 2025' - round_match = re.search(r'Round\s+\d+', tournament_text) - if round_match: - match_data['tournament'] += f" - {round_match.group(0)}" - else: - match_data['tournament'] = tournament_text - else: - # Fallback to text search - text = parent.get_text() - if 'TI2025' in text: - match_data['tournament'] = 'The International 2025' - round_match = re.search(r'Round\s+\d+', text) - if round_match: - match_data['tournament'] += f" - {round_match.group(0)}" - elif 'Major' in text: - major_match = re.search(r'[\w\s]+Major', text) - if major_match: - match_data['tournament'] = major_match.group(0).strip() - - # Mark if has score and if completed - if has_score: - # TBD vs TBD matches should NEVER be marked as having a score or completed - if match_data.get('team1') == 'TBD' and match_data.get('team2') == 'TBD': - has_score = False - match_data['completed'] = False - match_data['has_score'] = False - # Remove any incorrectly parsed score - if 'score' in match_data: - del match_data['score'] - else: - # Score already set above, extract score values - score_parts = re.match(r'(\d+)-(\d+)', match_data['score']) - if score_parts: - score1 = int(score_parts.group(1)) - score2 = int(score_parts.group(2)) - else: - score1 = score2 = 0 + 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 actually completed based on format - series_completed = False - if 'format' in match_data: - if 'Bo3' in match_data['format']: - # Bo3 is complete when someone reaches 2 wins - series_completed = (score1 >= 2 or score2 >= 2) - elif 'Bo5' in match_data['format']: - # Bo5 is complete when someone reaches 3 wins - series_completed = (score1 >= 3 or score2 >= 3) - elif 'Bo1' in match_data['format']: - # Bo1 is complete when there's any score - series_completed = True - else: - # Unknown format, assume completed if there's a score - series_completed = True - else: - # No format info, try to guess from score - # If someone has 2+ wins, likely a completed Bo3/Bo5 - series_completed = (score1 >= 2 or score2 >= 2) + # Check if series is completed + completed = self._is_series_completed(score1, score2, format_str) - match_data['completed'] = series_completed - match_data['has_score'] = True # Mark that there's a score even if not completed - - # Determine winner only if completed - if series_completed: - if score1 > score2: - match_data['winner'] = match_data.get('team1', 'Unknown') - else: - match_data['winner'] = match_data.get('team2', 'Unknown') - else: - match_data['completed'] = False - match_data['has_score'] = False + 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 + ) - # Generate ID if we have valid data - if 'team1' in match_data and 'team2' in match_data: - match_data['id'] = self._generate_match_id(match_data) - return match_data - except Exception as e: - pass + 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 _clean_team_name(self, name): + 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) @@ -289,50 +362,42 @@ class Dota2CalendarSync: name = re.sub(r'^\w+\s+\d+,\s+\d{4}.*', '', name).strip() return name - def _generate_match_id(self, match_data): + def _generate_match_id(self, team1: str, team2: str, tournament: Optional[str], match_datetime: datetime) -> str: """Generate a unique ID for a match""" - # Use teams and tournament for ID (not datetime to handle reschedules) id_parts = [] # For TBD vs TBD matches, include datetime to make them unique - if match_data.get('team1') == 'TBD' and match_data.get('team2') == 'TBD': - # Include datetime for TBD matches to avoid duplicates - if 'datetime' in match_data: - id_parts.append(str(match_data['datetime'])) - if 'tournament' in match_data: - id_parts.append(match_data['tournament']) + 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 - if 'team1' in match_data: - id_parts.append(match_data['team1']) - if 'team2' in match_data: - id_parts.append(match_data['team2']) - if 'tournament' in match_data: - id_parts.append(match_data['tournament']) + id_parts.extend([team1, team2]) + if tournament: + id_parts.append(tournament) else: - # Fall back to date if no tournament - if 'datetime' in match_data: - id_parts.append(str(match_data['datetime'].date())) + 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): + def _remove_duplicates(self, matches: List[Match]) -> List[Match]: """Remove duplicate matches based on ID""" unique_matches = {} for match in matches: - if match.get('id'): - unique_matches[match['id']] = match + unique_matches[match.id] = match return list(unique_matches.values()) - def get_existing_events(self, days_back=7, days_ahead=30): + @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' - print(f"Checking existing events in calendar...") + logger.info("Checking existing events in calendar...") events_result = self.service.events().list( calendarId=self.calendar_id, @@ -346,115 +411,138 @@ class Dota2CalendarSync: events = events_result.get('items', []) # Filter for Dota 2 events - # Build multiple indexes for better matching dota_events_by_id = {} dota_events_by_match = {} for event in events: summary = event.get('summary', '') - # Check for Dota events - old format has "Dota 2", new format has tournament brackets is_dota = ('Dota 2' in summary or 'The International' in summary or 'TI2025' in summary or - '[' in summary and 'vs' in summary) # New format has brackets + '[' in summary and 'vs' in summary) if is_dota: description = event.get('description', '') - # Extract ID from description (for old events) + # 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 - # Also create key based on teams and tournament for matching + # Create key based on teams and tournament summary = event.get('summary', '') - # Remove completed markers and scores summary = summary.replace('[COMPLETED] ', '') - # Remove checkmark and score (format: "✓ 2-1 Team vs Team") summary = re.sub(r'^✓\s+\d+[-:]\d+\s+', '', summary) - # Also handle old format with score at end + summary = re.sub(r'^\d+[-:]\d+\s+', '', summary) summary = re.sub(r'\s*\([0-9\-\?]+\)\s*$', '', summary) - # Try new format first: "Team1 vs Team2 [Tournament]" + # Extract teams and tournament match = re.search(r'^(.*?)\s+vs\s+(.*?)\s*\[(.*?)\]$', summary) - if match: - team1 = match.group(1).strip() - team2 = match.group(2).strip() - tournament = match.group(3).strip() - else: - # Try old format: "Dota 2 - Tournament: Team1 vs Team2" + 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() - if match: - - # Create match key (teams + tournament) - match_key = f"{team1}_{team2}_{tournament}" - dota_events_by_match[match_key] = event + match_key = f"{team1}_{team2}_{tournament}" + dota_events_by_match[match_key] = event - print(f"✓ Found {len(dota_events_by_id)} existing Dota 2 events") + logger.info(f"✓ Found {len(dota_events_by_id)} existing Dota 2 events") - # Return combined dictionary for backward compatibility combined = {} combined.update(dota_events_by_id) - # Store the by_match index as a special key combined['_by_match'] = dota_events_by_match return combined except Exception as e: - print(f"✗ Error fetching calendar events: {e}") - return {} + logger.error(f"✗ Error fetching calendar events: {e}") + raise - def create_calendar_event(self, match_data): - """Create a Google Calendar event for a match""" - team1 = match_data.get('team1', 'TBD') - team2 = match_data.get('team2', 'TBD') - tournament = match_data.get('tournament', '') + 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] - # New format: Teams first, tournament in brackets - if tournament: - summary = f"{team1} vs {team2} [{tournament}]" + # 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"{team1} vs {team2}" + summary = f"{match.team1} vs {match.team2}" # Build description description_parts = [] - if tournament: - description_parts.append(f"Tournament: {tournament}") - description_parts.append(f"Match: {team1} vs {team2}") - if 'format' in match_data: - description_parts.append(f"Format: {match_data['format']}") - if match_data.get('completed'): - description_parts.append(f"\n🏆 RESULT: {match_data.get('score', 'Unknown')}") - description_parts.append(f"Winner: {match_data.get('winner', 'Unknown')}") - description_parts.append(f"ID: {match_data['id']}") + 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 start and end times - start_time = match_data.get('datetime', datetime.now(pytz.UTC)) - # Estimate duration + # Set duration based on format duration = 2 - if 'format' in match_data: - if 'Bo5' in match_data['format']: + if match.format: + if 'Bo5' in match.format: duration = 4 - elif 'Bo3' in match_data['format']: + elif 'Bo3' in match.format: duration = 3 - elif 'Bo1' in match_data['format']: + elif 'Bo1' in match.format: duration = 1 - end_time = start_time + timedelta(hours=duration) + end_time = match.datetime + timedelta(hours=duration) event = { 'summary': summary, 'description': description, 'start': { - 'dateTime': start_time.isoformat(), + 'dateTime': match.datetime.isoformat(), 'timeZone': 'UTC', }, 'end': { @@ -472,7 +560,8 @@ class Dota2CalendarSync: return event - def update_event_time(self, event_id, new_datetime): + @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 @@ -490,22 +579,17 @@ class Dota2CalendarSync: event['start']['dateTime'] = new_datetime.isoformat() event['end']['dateTime'] = (new_datetime + duration).isoformat() - # Add note about time change to description + # 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: - # Update the timestamp - description = re.sub( - r'Last updated:.*', - f"Last updated: {datetime.now(pytz.UTC).strftime('%Y-%m-%d %H:%M UTC')}", - description - ) + description = re.sub(r'Last updated:.*', f"Last updated: {timestamp}", description) else: - # Add timestamp - description += f"\nLast updated: {datetime.now(pytz.UTC).strftime('%Y-%m-%d %H:%M UTC')}" + description += f"\nLast updated: {timestamp}" event['description'] = description # Update the event - updated_event = self.service.events().update( + self.service.events().update( calendarId=self.calendar_id, eventId=event_id, body=event @@ -514,10 +598,11 @@ class Dota2CalendarSync: return True except Exception as e: - print(f"Error updating event time: {e}") + logger.error(f"Error updating event time for {event_id}: {e}") return False - def update_event_with_teams(self, event_id, match_data): + @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 @@ -526,47 +611,30 @@ class Dota2CalendarSync: eventId=event_id ).execute() - team1 = match_data.get('team1', 'TBD') - team2 = match_data.get('team2', 'TBD') - tournament = match_data.get('tournament', '') - - # Update summary with actual team names - if tournament: - new_summary = f"{team1} vs {team2} [{tournament}]" + # Update summary + if match.tournament: + new_summary = f"{match.team1} vs {match.team2} [{match.tournament}]" else: - new_summary = f"{team1} vs {team2}" + new_summary = f"{match.team1} vs {match.team2}" # Update description description = event.get('description', '') - # Update the Match line - description = re.sub( - r'Match: .*?\n', - f"Match: {team1} vs {team2}\n", - description - ) - # Update the ID to the new one + 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_data.get('id', '')}", - 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: {datetime.now(pytz.UTC).strftime('%Y-%m-%d %H:%M UTC')}", - description - ) + description = re.sub(r'Teams updated:.*', f"Teams updated: {timestamp}", description) else: - description = description.replace('\nSource:', f"\nTeams updated: {datetime.now(pytz.UTC).strftime('%Y-%m-%d %H:%M UTC')}\nSource:") + description = description.replace('\nSource:', f"\nTeams updated: {timestamp}\nSource:") event['summary'] = new_summary event['description'] = description # Update the event - updated_event = self.service.events().update( + self.service.events().update( calendarId=self.calendar_id, eventId=event_id, body=event @@ -575,10 +643,11 @@ class Dota2CalendarSync: return True except Exception as e: - print(f"Error updating event with teams: {e}") + logger.error(f"Error updating event with teams for {event_id}: {e}") return False - def update_event_with_score(self, event_id, match_data): + @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 @@ -587,38 +656,28 @@ class Dota2CalendarSync: eventId=event_id ).execute() - # Update the description with current score + # Update description description = event.get('description', '') + score_text = f"📊 CURRENT SCORE: {match.score or 'Unknown'}" - # Check if score is already in the description if '📊 CURRENT SCORE:' in description: - # Update existing score - description = re.sub( - r'📊 CURRENT SCORE:.*?\n', - f"📊 CURRENT SCORE: {match_data.get('score', 'Unknown')}\n", - description - ) + description = re.sub(r'📊 CURRENT SCORE:.*?\n', f"{score_text}\n", description) else: - # Add new score - score_text = f"\n📊 CURRENT SCORE: {match_data.get('score', 'Unknown')}\n" if 'ID:' in description: - description = description.replace('ID:', score_text + 'ID:') + description = description.replace('ID:', f"\n{score_text}\nID:") else: - description += score_text + description += f"\n{score_text}" - # Update the summary to show current score (without checkmark) + # Update summary summary = event.get('summary', '') - # Remove any existing score (including multiple scores) summary = re.sub(r'^(\d+[-:]\d+\s+)+', '', summary) - # Add new score at the beginning - score = match_data.get('score', '?-?') - summary = f"{score} {summary}" + summary = f"{match.score} {summary}" - # Update the event event['description'] = description event['summary'] = summary - updated_event = self.service.events().update( + # Update the event + self.service.events().update( calendarId=self.calendar_id, eventId=event_id, body=event @@ -627,10 +686,11 @@ class Dota2CalendarSync: return True except Exception as e: - print(f"Error updating event score: {e}") + logger.error(f"Error updating event score for {event_id}: {e}") return False - def update_event_with_result(self, event_id, match_data): + @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 @@ -639,41 +699,34 @@ class Dota2CalendarSync: eventId=event_id ).execute() - # Update the description with results + # Update description description = event.get('description', '') + result_text = f"🏆 RESULT: {match.score or 'Unknown'}\nWinner: {match.winner or 'Unknown'}" - # Check if results are already in the description if '🏆 RESULT:' in description: - # Update existing result description = re.sub( r'🏆 RESULT:.*?\n.*?Winner:.*?\n', - f"🏆 RESULT: {match_data.get('score', 'Unknown')}\nWinner: {match_data.get('winner', 'Unknown')}\n", + f"{result_text}\n", description, flags=re.DOTALL ) else: - # Add new result - result_text = f"\n🏆 RESULT: {match_data.get('score', 'Unknown')}\nWinner: {match_data.get('winner', 'Unknown')}\n" if 'ID:' in description: - description = description.replace('ID:', result_text + 'ID:') + description = description.replace('ID:', f"\n{result_text}\nID:") else: - description += result_text + description += f"\n{result_text}" - # Update the summary to show it's completed with result + # Update summary summary = event.get('summary', '') - # First remove any existing scores (in-progress scores) summary = re.sub(r'^(\d+[-:]\d+\s+)+', '', summary) - # Remove any existing checkmark summary = re.sub(r'^✓\s+', '', summary) - # Add checkmark and final score - score = match_data.get('score', '?-?') - summary = f"✓ {score} {summary}" + summary = f"✓ {match.score} {summary}" - # Update the event event['description'] = description event['summary'] = summary - updated_event = self.service.events().update( + # Update the event + self.service.events().update( calendarId=self.calendar_id, eventId=event_id, body=event @@ -682,10 +735,10 @@ class Dota2CalendarSync: return True except Exception as e: - print(f"Error updating event: {e}") + logger.error(f"Error updating event result for {event_id}: {e}") return False - def check_time_difference(self, event_datetime, new_datetime): + 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): @@ -705,9 +758,10 @@ class Dota2CalendarSync: # Calculate difference in minutes diff = abs((event_dt - new_datetime).total_seconds() / 60) - return diff >= 5 # Return True if difference is 5 minutes or more + return diff >= 5 - def delete_calendar_event(self, event_id): + @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( @@ -716,470 +770,352 @@ class Dota2CalendarSync: ).execute() return True except Exception as e: - print(f"Error deleting event: {e}") + logger.error(f"Error deleting event {event_id}: {e}") return False - def sync_matches_to_calendar(self, dry_run=False, update_results=True, update_times=True, delete_old_tbd=True): - """Main sync function with time updates""" - print("\n" + "="*50) - print("Starting Dota 2 Calendar Sync v3") - print("="*50 + "\n") + 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 + } - # Fetch all matches - upcoming_matches, completed_matches = self.fetch_all_matches() - - if not upcoming_matches and not completed_matches: - print("No matches found to sync") - return - - # Get existing events - existing_events = self.get_existing_events(days_back=7, days_ahead=30) - - # Counters - added_count = 0 - skipped_count = 0 - updated_count = 0 - time_updated_count = 0 - error_count = 0 - deleted_tbd_count = 0 - - # Track which TBD events were updated updated_tbd_events = set() - - # Process upcoming matches - print("\nProcessing upcoming matches...") - print("-" * 30) - now = datetime.now(pytz.UTC) - # Include matches from the last 12 hours (to catch ongoing matches with scores) twelve_hours_ago = now - timedelta(hours=12) - future_matches = [m for m in upcoming_matches - if m.get('datetime', now) >= twelve_hours_ago] + + 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: - match_id = match.get('id') - team1 = match.get('team1', 'TBD') - team2 = match.get('team2', 'TBD') - match_time = match.get('datetime', now) - tournament = match.get('tournament', '') - - if not match_id: - continue - - # Try to find existing event by ID or by match details - existing_event = None - - # First try by ID - if match_id in existing_events: - existing_event = existing_events[match_id] - - # If not found, try by match details (teams + tournament) - if not existing_event and '_by_match' in existing_events: - match_key = f"{team1}_{team2}_{tournament}" - if match_key in existing_events['_by_match']: - existing_event = existing_events['_by_match'][match_key] + try: + existing_event = self.find_existing_event(match, existing_events) - # Also try to find by teams only (ignoring score) for live updates - # This handles cases where score changes during match - if not existing_event: - for event_key, event in existing_events['_by_match'].items(): - # Check if teams match (order independent) - if (f"{team1}_{team2}" in event_key or f"{team2}_{team1}" in event_key) and tournament in event_key: - existing_event = event - print(f" → Found existing match by teams: {team1} vs {team2}") - break - - # Special handling for TBD matches that might have been updated - # Look for TBD events at the same time and tournament - if not existing_event and '_by_match' in existing_events: - # Only look for TBD to update if current match is NOT TBD vs TBD - # (we don't want to match TBD vs TBD with other TBD vs TBD) - if not (team1 == 'TBD' and team2 == 'TBD'): - # Check if this match used to be TBD - for event_key, event in existing_events['_by_match'].items(): - if 'TBD_TBD' in event_key and 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')) - # Relaxed time matching: within 1 hour (3600 seconds) - if abs((event_dt - match_time).total_seconds()) < 3600: # Within 1 hour - existing_event = event - print(f" → Found TBD match to update: {team1} vs {team2}") - print(f" Time difference: {abs((event_dt - match_time).total_seconds())/60:.0f} minutes") - break - - 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 (team1 != 'TBD' or team2 != 'TBD') - - if is_tbd_update: - # Update TBD match with actual teams - if dry_run: - print(f"◯ Would update TBD match with teams: {team1} vs {team2}") - updated_count += 1 - else: - if self.update_event_with_teams(existing_event['id'], match): - print(f"✓ Updated TBD match with teams: {team1} vs {team2}") - updated_count += 1 - updated_tbd_events.add(existing_event['id']) - time.sleep(0.2) - else: - print(f"✗ Failed to update TBD match: {team1} vs {team2}") - error_count += 1 - # Check if this match has a score (completed or in-progress) and needs update - elif match.get('has_score') and update_results: - # Check current event status + if existing_event: + # Check if this is a TBD match that now has team names summary = existing_event.get('summary', '') - description = existing_event.get('description', '') - current_score = None + is_tbd_update = 'TBD' in summary and (match.team1 != 'TBD' or match.team2 != 'TBD') - # Try to extract current score from summary - score_in_summary = re.search(r'✓?\s*(\d+[-:]\d+)', summary) - if score_in_summary: - current_score = score_in_summary.group(1).replace(':', '-') - - # Check if score needs update - new_score = match.get('score', 'Unknown') - - if current_score == new_score: - print(f"⊘ Score unchanged: {team1} vs {team2} ({new_score})") - skipped_count += 1 - else: - if match.get('completed'): - # Series is completed - if dry_run: - print(f"◯ Would update completed result: {team1} vs {team2} - {new_score}") - updated_count += 1 - else: - if self.update_event_with_result(existing_event['id'], match): - print(f"✓ Updated completed result: {team1} vs {team2} - {new_score}") - updated_count += 1 - time.sleep(0.2) - else: - print(f"✗ Failed to update: {team1} vs {team2}") - error_count += 1 - else: - # Series is in-progress with partial score - if dry_run: - print(f"◯ Would update in-progress score: {team1} vs {team2} - {new_score}") - updated_count += 1 - else: - if self.update_event_with_score(existing_event['id'], match): - print(f"📊 Updated in-progress score: {team1} vs {team2} - {new_score}") - updated_count += 1 - time.sleep(0.2) - else: - print(f"✗ Failed to update score: {team1} vs {team2}") - error_count += 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_time): - # Time has changed + if is_tbd_update: if dry_run: - old_time = datetime.fromisoformat(event_start.replace('Z', '+00:00')) - print(f"◯ Would update time: {team1} vs {team2}") - print(f" Old: {old_time.strftime('%Y-%m-%d %H:%M UTC')}") - print(f" New: {match_time.strftime('%Y-%m-%d %H:%M UTC')}") - time_updated_count += 1 + logger.info(f"◯ Would update TBD match with teams: {match.team1} vs {match.team2}") + counters['updated'] += 1 else: - old_time = datetime.fromisoformat(event_start.replace('Z', '+00:00')) - if self.update_event_time(existing_event['id'], match_time): - print(f"⏰ Updated time: {team1} vs {team2}") - print(f" Old: {old_time.strftime('%Y-%m-%d %H:%M UTC')}") - print(f" New: {match_time.strftime('%Y-%m-%d %H:%M UTC')}") - time_updated_count += 1 + 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: - print(f"✗ Failed to update time: {team1} vs {team2}") - error_count += 1 - else: - print(f"⊘ No change: {team1} vs {team2}") - skipped_count += 1 - else: - print(f"⊘ Skipping (exists): {team1} vs {team2}") - skipped_count += 1 - else: - # New match - if dry_run: - print(f"◯ Would add: {team1} vs {team2} at {match_time.strftime('%Y-%m-%d %H:%M UTC')}") - added_count += 1 - else: - try: - event = self.create_calendar_event(match) - self.service.events().insert( - calendarId=self.calendar_id, - body=event - ).execute() - print(f"✓ Added: {team1} vs {team2} at {match_time.strftime('%Y-%m-%d %H:%M UTC')}") - added_count += 1 - time.sleep(0.2) - except Exception as e: - print(f"✗ Error adding {team1} vs {team2}: {e}") - error_count += 1 - - # Process completed matches for results - if update_results and completed_matches: - print("\nProcessing completed match results...") - print("-" * 30) - - for match in completed_matches: - match_id = match.get('id') - team1 = match.get('team1', 'TBD') - team2 = match.get('team2', 'TBD') - score = match.get('score', 'Unknown') - tournament = match.get('tournament', '') - - if not match_id: - continue - - # Try to find existing event by ID or by match details - existing_event = None - - # First try by ID - if match_id in existing_events: - existing_event = existing_events[match_id] - - # If not found, try by match details - if not existing_event and '_by_match' in existing_events: - match_key = f"{team1}_{team2}_{tournament}" - if match_key in existing_events['_by_match']: - existing_event = existing_events['_by_match'][match_key] + counters['errors'] += 1 - # Also try to find by teams only (for live score updates) - if not existing_event: - for event_key, event in existing_events['_by_match'].items(): - # Check if teams match (order independent) and tournament matches - if (f"{team1}_{team2}" in event_key or f"{team2}_{team1}" in event_key) and tournament in event_key: - existing_event = event - print(f" → Found existing match by teams: {team1} vs {team2}") - break + # 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: - print(f"⊘ Already updated: {team1} vs {team2} ({score})") + logger.info(f"⊘ Already updated: {match.team1} vs {match.team2} ({match.score})") else: if dry_run: - print(f"◯ Would update result: {team1} vs {team2} - {score}") - updated_count += 1 + 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): - print(f"✓ Updated result: {team1} vs {team2} - {score}") - updated_count += 1 + logger.info(f"✓ Updated result: {match.team1} vs {match.team2} - {match.score}") + counters['updated'] += 1 time.sleep(0.2) else: - print(f"✗ Failed to update: {team1} vs {team2}") - error_count += 1 + counters['errors'] += 1 + + except Exception as e: + logger.error(f"Error processing completed match {match.team1} vs {match.team2}: {e}") + counters['errors'] += 1 + continue - # Delete old TBD events that are past and not updated - # Also check for duplicate TBD events at the same time - # Also delete TBD events when a confirmed match exists at the same time - if delete_old_tbd and not dry_run: - print("\nChecking for expired, duplicate, and superseded TBD events to delete...") - print("-" * 30) + 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 - # Group all events by time to find duplicates and superseded TBD events - events_by_time = {} - tbd_by_time = {} + summary = event.get('summary', '') + event_id = event['id'] - # Get all events and group 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) - # 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 for checking if match has ended - 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: - # If no end time, assume 3 hours duration - 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 (end time has passed) and still contains TBD - if event_end_dt < now: + # 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): - print(f"🗑️ Deleted ended TBD event: {summary} (ended at {event_end_dt.strftime('%Y-%m-%d %H:%M UTC')})") - deleted_tbd_count += 1 + logger.info(f"🗑️ Deleted ended TBD event: {summary}") + deleted_count += 1 time.sleep(0.2) - else: - print(f"✗ Failed to delete ended TBD event: {summary}") - error_count += 1 - continue # Don't process this event further - - # 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 that have been superseded by confirmed matches - for time_key, events in events_by_time.items(): - if events['confirmed'] and events['tbd']: - # We have both confirmed and TBD events at the same time - for tbd_event in events['tbd']: - tbd_summary = tbd_event.get('summary', '') - - # Check if this is a complete TBD vs TBD event - if 'TBD vs TBD' in tbd_summary: - # Delete TBD vs TBD when there's any confirmed match at the same time - # Since TBD vs TBD is a placeholder, any confirmed match supersedes it - if events['confirmed']: - confirmed_event = events['confirmed'][0] # Use first confirmed match for logging - confirmed_summary = confirmed_event.get('summary', '') - if self.delete_calendar_event(tbd_event['id']): - print(f"🗑️ Deleted TBD vs TBD event at same time as confirmed match") - print(f" TBD event: {tbd_summary}") - print(f" Confirmed match: {confirmed_summary}") - deleted_tbd_count += 1 - time.sleep(0.2) - else: - print(f"✗ Failed to delete TBD vs TBD event: {tbd_summary}") - error_count += 1 - else: - # For partial TBD events (one team is known), check for team match - team_match = re.search(r'(\w+)\s+vs\s+TBD|TBD\s+vs\s+(\w+)', tbd_summary) - if team_match: - team_in_tbd = team_match.group(1) or team_match.group(2) - - # Check if this team has a confirmed match - for confirmed_event in events['confirmed']: - confirmed_summary = confirmed_event.get('summary', '') - if team_in_tbd and team_in_tbd in confirmed_summary: - # This TBD event has been superseded - if self.delete_calendar_event(tbd_event['id']): - print(f"🗑️ Deleted superseded TBD event: {tbd_summary}") - print(f" Replaced by: {confirmed_summary}") - deleted_tbd_count += 1 - time.sleep(0.2) - else: - print(f"✗ Failed to delete TBD event: {tbd_summary}") - error_count += 1 - break - - # Delete duplicate TBD vs TBD events at the same time - for time_key, events in tbd_by_time.items(): - if len(events) > 1: - print(f"Found {len(events)} duplicate TBD events at {time_key}") - # Keep the first one, delete the rest - for event in events[1:]: - if self.delete_calendar_event(event['id']): - print(f"🗑️ Deleted duplicate TBD event: {event['summary']}") - deleted_tbd_count += 1 - time.sleep(0.2) - else: - print(f"✗ Failed to delete duplicate TBD event: {event['summary']}") - error_count += 1 - - # Also check for duplicate matches with different completion states - # Group matches by teams and date (not exact time) - matches_by_teams_date = {} - for key, event in existing_events.items(): - if key == '_by_match': + else: + logger.info(f"◯ Would delete ended TBD event: {summary}") + deleted_count += 1 continue - summary = event.get('summary', '') - # Skip TBD matches - if 'TBD' in summary: - 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', '') - # Extract teams from summary - # Remove completion markers and scores - clean_summary = re.sub(r'^✓\s*\d+[-:]\d+\s*', '', summary) - clean_summary = re.sub(r'^\d+[-:]\d+\s*', '', clean_summary) - - # Extract teams - teams_match = re.search(r'([\w\s]+)\s+vs\s+([\w\s]+)\s*\[', clean_summary) - if teams_match: - team1 = teams_match.group(1).strip() - team2 = teams_match.group(2).strip() - - # Get date - event_start = event['start'].get('dateTime', event['start'].get('date')) - event_dt = datetime.fromisoformat(event_start.replace('Z', '+00:00')) - date_key = event_dt.strftime('%Y-%m-%d') - - # Create key for this match - match_key = f"{min(team1, team2)}_vs_{max(team1, team2)}_{date_key}" - - if match_key not in matches_by_teams_date: - matches_by_teams_date[match_key] = [] - matches_by_teams_date[match_key].append(event) - - # Delete duplicates, keeping the completed one - for match_key, events in matches_by_teams_date.items(): - if len(events) > 1: - # Sort by completion status (completed first) and time - def sort_key(e): - summary = e.get('summary', '') - is_completed = '✓' in summary - event_start = e['start'].get('dateTime', e['start'].get('date')) - return (not is_completed, event_start) # Completed first, then by time - - sorted_events = sorted(events, key=sort_key) - - # Keep the first (preferably completed) event - print(f"Found {len(events)} duplicate matches: {sorted_events[0]['summary']}") - for event in sorted_events[1:]: - if self.delete_calendar_event(event['id']): - print(f"🗑️ Deleted duplicate match: {event['summary']}") - deleted_tbd_count += 1 - time.sleep(0.2) + 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: - print(f"✗ Failed to delete duplicate: {event['summary']}") - error_count += 1 + logger.info(f"◯ Would delete TBD vs TBD event: {tbd_summary}") + deleted_count += 1 - # Summary - print("\n" + "="*50) - print("Sync Summary") - print("="*50) - print(f"✓ Added: {added_count} matches") - if time_updated_count > 0: - print(f"⏰ Time updated: {time_updated_count} matches") - if updated_count > 0: - print(f"✓ Results updated: {updated_count} matches") - if deleted_tbd_count > 0: - print(f"🗑️ Deleted: {deleted_tbd_count} expired TBD events") - print(f"⊘ Skipped: {skipped_count} matches (no changes)") - if error_count > 0: - print(f"✗ Errors: {error_count} matches") + # 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 - if dry_run: - print("\n⚠ DRY RUN - No actual changes were made") + 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") - print("\n✓ Sync complete!") + 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 v3' + description='Sync Dota 2 Tier 1 matches from Liquipedia to Google Calendar v4.0' ) parser.add_argument( '--calendar-id', @@ -1206,15 +1142,24 @@ def main(): 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 - print("\n" + "!"*60) - print("Dota 2 Calendar Sync v3") - print("Features: Match sync, result updates, time change detection") - print("Service Account: calendar-bot@tunpok.iam.gserviceaccount.com") - print("!"*60 + "\n") + 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: @@ -1231,10 +1176,10 @@ def main(): ) except KeyboardInterrupt: - print("\n\nSync cancelled by user") + logger.info("\n\nSync cancelled by user") sys.exit(0) except Exception as e: - print(f"\n✗ Fatal error: {e}") + logger.error(f"\n✗ Fatal error: {e}") sys.exit(1) if __name__ == "__main__":