docs: 为 v4.0 版本添加详细的方法注释

- 为所有公共方法添加完整的 docstring
- 添加参数说明 (Args) 和返回值说明 (Returns)
- 为复杂逻辑添加实现细节说明
- 改进 Match dataclass 的文档
- 添加装饰器和辅助函数的使用示例

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Ching L 2025-09-12 18:11:52 +08:00
parent aa892e17fb
commit a790fc8489

View File

@ -44,7 +44,20 @@ class MatchFormat(Enum):
@dataclass @dataclass
class Match: class Match:
"""Data class for match information""" """Data class for match information
Attributes:
id: Unique identifier for the match (MD5 hash)
team1: Name of the first team
team2: Name of the second team
datetime: Match scheduled datetime (timezone aware)
tournament: Tournament name (e.g., 'The International 2025')
format: Match format (Bo1, Bo3, Bo5)
score: Current or final score (e.g., '2-1')
completed: Whether the series is completed
has_score: Whether the match has any score (including partial)
winner: Name of the winning team if completed
"""
id: str id: str
team1: str team1: str
team2: str team2: str
@ -57,12 +70,20 @@ class Match:
winner: Optional[str] = None winner: Optional[str] = None
def __post_init__(self): def __post_init__(self):
"""Ensure datetime is timezone aware""" """Post-initialization processing
Ensures datetime is always timezone aware (UTC)
"""
if self.datetime and self.datetime.tzinfo is None: if self.datetime and self.datetime.tzinfo is None:
self.datetime = self.datetime.replace(tzinfo=pytz.UTC) self.datetime = self.datetime.replace(tzinfo=pytz.UTC)
def to_dict(self) -> Dict[str, Any]: def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for compatibility""" """Convert Match object to dictionary
Returns:
Dict containing all match attributes
Used for backward compatibility with existing code
"""
return { return {
'id': self.id, 'id': self.id,
'team1': self.team1, 'team1': self.team1,
@ -77,7 +98,21 @@ class Match:
} }
def retry_on_exception(max_retries: int = 3, delay: float = 1.0, backoff: float = 2.0): def retry_on_exception(max_retries: int = 3, delay: float = 1.0, backoff: float = 2.0):
"""Decorator for retrying functions with exponential backoff""" """Decorator for retrying functions with exponential backoff
Args:
max_retries: Maximum number of retry attempts
delay: Initial delay between retries in seconds
backoff: Multiplier for delay after each retry (exponential backoff)
Returns:
Decorated function that automatically retries on exceptions
Example:
@retry_on_exception(max_retries=3, delay=2.0)
def fetch_data():
# This will retry up to 3 times with delays of 2s, 4s, 8s
"""
def decorator(func): def decorator(func):
@wraps(func) @wraps(func)
def wrapper(*args, **kwargs): def wrapper(*args, **kwargs):
@ -101,13 +136,32 @@ def retry_on_exception(max_retries: int = 3, delay: float = 1.0, backoff: float
return decorator return decorator
class Dota2CalendarSync: class Dota2CalendarSync:
"""Main class for synchronizing Dota 2 matches to Google Calendar
Handles fetching matches from Liquipedia and syncing them to Google Calendar,
including creating new events, updating scores, and managing TBD placeholders.
"""
def __init__(self, credentials_file='credentials.json', calendar_id='primary'): def __init__(self, credentials_file='credentials.json', calendar_id='primary'):
"""Initialize the sync service
Args:
credentials_file: Path to Google service account credentials JSON
calendar_id: Google Calendar ID or 'primary' for main calendar
"""
self.credentials_file = credentials_file self.credentials_file = credentials_file
self.calendar_id = calendar_id self.calendar_id = calendar_id
self.service = self._authenticate() self.service = self._authenticate()
def _authenticate(self): def _authenticate(self):
"""Authenticate with Google Calendar using service account credentials""" """Authenticate with Google Calendar using service account credentials
Returns:
Google Calendar service object
Raises:
SystemExit: If authentication fails
"""
try: try:
credentials = service_account.Credentials.from_service_account_file( credentials = service_account.Credentials.from_service_account_file(
self.credentials_file, self.credentials_file,
@ -122,7 +176,14 @@ class Dota2CalendarSync:
@retry_on_exception(max_retries=3, delay=2.0) @retry_on_exception(max_retries=3, delay=2.0)
def fetch_all_matches(self) -> Tuple[List[Match], List[Match]]: def fetch_all_matches(self) -> Tuple[List[Match], List[Match]]:
"""Fetch both upcoming and completed matches from Liquipedia""" """Fetch both upcoming and completed matches from Liquipedia
Scrapes the Liquipedia matches page for Tier 1 Dota 2 tournaments.
Automatically retries on network failures.
Returns:
Tuple of (upcoming_matches, completed_matches) lists
"""
url = 'https://liquipedia.net/dota2/Liquipedia:Matches' url = 'https://liquipedia.net/dota2/Liquipedia:Matches'
headers = { headers = {
'User-Agent': 'Dota2CalendarSync/3.0 (https://github.com/youruser/dota2-calendar)' 'User-Agent': 'Dota2CalendarSync/3.0 (https://github.com/youruser/dota2-calendar)'
@ -175,7 +236,18 @@ class Dota2CalendarSync:
return [], [] return [], []
def _parse_match(self, parent, timestamp_elem) -> Optional[Match]: def _parse_match(self, parent, timestamp_elem) -> Optional[Match]:
"""Parse match data from an element using HTML structure""" """Parse match data from an HTML element
Extracts team names, scores, format, and tournament information
from Liquipedia's HTML structure.
Args:
parent: Parent div element containing match information
timestamp_elem: Span element with data-timestamp attribute
Returns:
Match object if parsing successful, None otherwise
"""
try: try:
# Get timestamp # Get timestamp
timestamp = timestamp_elem.get('data-timestamp') timestamp = timestamp_elem.get('data-timestamp')
@ -267,7 +339,17 @@ class Dota2CalendarSync:
return None return None
def _extract_score(self, parent) -> Optional[str]: def _extract_score(self, parent) -> Optional[str]:
"""Extract score from match element""" """Extract score from match element
Looks for score in structured HTML elements first,
then falls back to text pattern matching.
Args:
parent: Parent element containing match information
Returns:
Score string (e.g., '2-1') or None if no valid score found
"""
# Look for score in structured elements # Look for score in structured elements
score_holder = parent.find('div', class_='match-info-header-scoreholder') score_holder = parent.find('div', class_='match-info-header-scoreholder')
if score_holder: if score_holder:
@ -295,7 +377,14 @@ class Dota2CalendarSync:
return None return None
def _extract_format(self, parent) -> Optional[str]: def _extract_format(self, parent) -> Optional[str]:
"""Extract match format""" """Extract match format (Bo1, Bo3, Bo5)
Args:
parent: Parent element containing match information
Returns:
Format string (e.g., 'Bo3') or None if not found
"""
format_elem = parent.find('span', class_='match-info-header-scoreholder-lower') format_elem = parent.find('span', class_='match-info-header-scoreholder-lower')
if format_elem: if format_elem:
format_text = format_elem.get_text().strip() format_text = format_elem.get_text().strip()
@ -312,7 +401,16 @@ class Dota2CalendarSync:
return None return None
def _extract_tournament(self, parent) -> Optional[str]: def _extract_tournament(self, parent) -> Optional[str]:
"""Extract tournament name""" """Extract tournament name from match element
Special handling for TI2025 and Major tournaments.
Args:
parent: Parent element containing match information
Returns:
Tournament name or None if not found
"""
tournament_elem = parent.find('div', class_='match-info-tournament') tournament_elem = parent.find('div', class_='match-info-tournament')
if tournament_elem: if tournament_elem:
tournament_text = tournament_elem.get_text().strip() tournament_text = tournament_elem.get_text().strip()
@ -341,7 +439,16 @@ class Dota2CalendarSync:
return None return None
def _is_series_completed(self, score1: int, score2: int, format_str: Optional[str]) -> bool: 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""" """Check if a series is completed based on score and format
Args:
score1: Score for team 1
score2: Score for team 2
format_str: Match format (Bo1, Bo3, Bo5)
Returns:
True if series is completed (someone has won enough games)
"""
if not format_str: if not format_str:
return score1 >= 2 or score2 >= 2 return score1 >= 2 or score2 >= 2
@ -355,7 +462,16 @@ class Dota2CalendarSync:
return True return True
def _clean_team_name(self, name: str) -> str: def _clean_team_name(self, name: str) -> str:
"""Clean and normalize team name""" """Clean and normalize team name
Removes extra whitespace, parenthetical notes, dates, etc.
Args:
name: Raw team name from HTML
Returns:
Cleaned team name
"""
name = re.sub(r'\s+', ' ', name).strip() name = re.sub(r'\s+', ' ', name).strip()
name = re.sub(r'\s*\(.*?\)\s*$', '', name) name = re.sub(r'\s*\(.*?\)\s*$', '', name)
name = re.sub(r'^\d{4}-\d{2}-\d{2}.*', '', name).strip() name = re.sub(r'^\d{4}-\d{2}-\d{2}.*', '', name).strip()
@ -363,7 +479,20 @@ class Dota2CalendarSync:
return name return name
def _generate_match_id(self, team1: str, team2: str, tournament: Optional[str], match_datetime: datetime) -> str: def _generate_match_id(self, team1: str, team2: str, tournament: Optional[str], match_datetime: datetime) -> str:
"""Generate a unique ID for a match""" """Generate a unique ID for a match
Uses MD5 hash of team names and tournament. For TBD vs TBD matches,
includes datetime to ensure uniqueness.
Args:
team1: First team name
team2: Second team name
tournament: Tournament name
match_datetime: Match scheduled time
Returns:
16-character hex string ID
"""
id_parts = [] id_parts = []
# For TBD vs TBD matches, include datetime to make them unique # For TBD vs TBD matches, include datetime to make them unique
@ -383,7 +512,14 @@ class Dota2CalendarSync:
return hashlib.md5(unique_string.encode()).hexdigest()[:16] return hashlib.md5(unique_string.encode()).hexdigest()[:16]
def _remove_duplicates(self, matches: List[Match]) -> List[Match]: def _remove_duplicates(self, matches: List[Match]) -> List[Match]:
"""Remove duplicate matches based on ID""" """Remove duplicate matches based on ID
Args:
matches: List of Match objects
Returns:
List with duplicates removed (keeps last occurrence)
"""
unique_matches = {} unique_matches = {}
for match in matches: for match in matches:
unique_matches[match.id] = match unique_matches[match.id] = match
@ -391,7 +527,21 @@ class Dota2CalendarSync:
@retry_on_exception(max_retries=3, delay=1.0) @retry_on_exception(max_retries=3, delay=1.0)
def get_existing_events(self, days_back=7, days_ahead=30) -> Dict[str, Any]: def get_existing_events(self, days_back=7, days_ahead=30) -> Dict[str, Any]:
"""Get existing Dota 2 events from Google Calendar""" """Get existing Dota 2 events from Google Calendar
Fetches events within the specified time range and builds two indexes:
- By match ID (from event description)
- By match details (teams + tournament)
Args:
days_back: Number of days to look back
days_ahead: Number of days to look ahead
Returns:
Dictionary with event indexes:
- Keys are match IDs
- Special '_by_match' key contains team+tournament index
"""
try: try:
now = datetime.utcnow() now = datetime.utcnow()
time_min = (now - timedelta(days=days_back)).isoformat() + 'Z' time_min = (now - timedelta(days=days_back)).isoformat() + 'Z'
@ -467,7 +617,21 @@ class Dota2CalendarSync:
raise raise
def find_existing_event(self, match: Match, existing_events: Dict[str, Any]) -> Optional[Dict[str, Any]]: def find_existing_event(self, match: Match, existing_events: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""Find existing calendar event for a match""" """Find existing calendar event for a match
Tries multiple matching strategies:
1. Direct ID match
2. Team + tournament match
3. Teams only (for live score updates)
4. TBD placeholder match (within 1 hour time window)
Args:
match: Match object to find
existing_events: Dictionary of existing calendar events
Returns:
Calendar event dict if found, None otherwise
"""
# Try by ID first # Try by ID first
if match.id in existing_events: if match.id in existing_events:
return existing_events[match.id] return existing_events[match.id]
@ -504,7 +668,14 @@ class Dota2CalendarSync:
return None return None
def create_calendar_event(self, match: Match) -> Dict[str, Any]: def create_calendar_event(self, match: Match) -> Dict[str, Any]:
"""Create a Google Calendar event for a match""" """Create a Google Calendar event for a match
Args:
match: Match object with event details
Returns:
Dictionary with Google Calendar event format
"""
# Build summary # Build summary
if match.tournament: if match.tournament:
summary = f"{match.team1} vs {match.team2} [{match.tournament}]" summary = f"{match.team1} vs {match.team2} [{match.tournament}]"
@ -562,7 +733,17 @@ class Dota2CalendarSync:
@retry_on_exception(max_retries=2, delay=1.0) @retry_on_exception(max_retries=2, delay=1.0)
def update_event_time(self, event_id: str, new_datetime: datetime) -> bool: def update_event_time(self, event_id: str, new_datetime: datetime) -> bool:
"""Update the time of an existing event""" """Update the time of an existing event
Preserves event duration when updating start time.
Args:
event_id: Google Calendar event ID
new_datetime: New start time for the event
Returns:
True if update successful, False otherwise
"""
try: try:
# Get the existing event # Get the existing event
event = self.service.events().get( event = self.service.events().get(
@ -603,7 +784,17 @@ class Dota2CalendarSync:
@retry_on_exception(max_retries=2, delay=1.0) @retry_on_exception(max_retries=2, delay=1.0)
def update_event_with_teams(self, event_id: str, match: Match) -> bool: def update_event_with_teams(self, event_id: str, match: Match) -> bool:
"""Update a TBD event with actual team names""" """Update a TBD event with actual team names
Used when TBD placeholder teams are determined.
Args:
event_id: Google Calendar event ID
match: Match object with updated team names
Returns:
True if update successful, False otherwise
"""
try: try:
# Get the existing event # Get the existing event
event = self.service.events().get( event = self.service.events().get(
@ -648,7 +839,17 @@ class Dota2CalendarSync:
@retry_on_exception(max_retries=2, delay=1.0) @retry_on_exception(max_retries=2, delay=1.0)
def update_event_with_score(self, event_id: str, match: Match) -> bool: def update_event_with_score(self, event_id: str, match: Match) -> bool:
"""Update an existing calendar event with in-progress score""" """Update an existing calendar event with in-progress score
Shows current score for ongoing matches (not yet completed).
Args:
event_id: Google Calendar event ID
match: Match object with current score
Returns:
True if update successful, False otherwise
"""
try: try:
# Get the existing event # Get the existing event
event = self.service.events().get( event = self.service.events().get(
@ -691,7 +892,17 @@ class Dota2CalendarSync:
@retry_on_exception(max_retries=2, delay=1.0) @retry_on_exception(max_retries=2, delay=1.0)
def update_event_with_result(self, event_id: str, match: Match) -> bool: def update_event_with_result(self, event_id: str, match: Match) -> bool:
"""Update an existing calendar event with match results""" """Update an existing calendar event with match results
Marks match as completed with final score and winner.
Args:
event_id: Google Calendar event ID
match: Match object with final results
Returns:
True if update successful, False otherwise
"""
try: try:
# Get the existing event # Get the existing event
event = self.service.events().get( event = self.service.events().get(
@ -739,7 +950,15 @@ class Dota2CalendarSync:
return False return False
def check_time_difference(self, event_datetime: Any, new_datetime: datetime) -> bool: def check_time_difference(self, event_datetime: Any, new_datetime: datetime) -> bool:
"""Check if there's a significant time difference (>= 5 minutes)""" """Check if there's a significant time difference (>= 5 minutes)
Args:
event_datetime: Current event datetime (string or datetime object)
new_datetime: New datetime to compare
Returns:
True if difference is 5 minutes or more
"""
# Parse event datetime # Parse event datetime
if isinstance(event_datetime, str): if isinstance(event_datetime, str):
if event_datetime.endswith('Z'): if event_datetime.endswith('Z'):
@ -762,7 +981,14 @@ class Dota2CalendarSync:
@retry_on_exception(max_retries=2, delay=1.0) @retry_on_exception(max_retries=2, delay=1.0)
def delete_calendar_event(self, event_id: str) -> bool: def delete_calendar_event(self, event_id: str) -> bool:
"""Delete a calendar event""" """Delete a calendar event
Args:
event_id: Google Calendar event ID to delete
Returns:
True if deletion successful, False otherwise
"""
try: try:
self.service.events().delete( self.service.events().delete(
calendarId=self.calendar_id, calendarId=self.calendar_id,
@ -775,7 +1001,24 @@ class Dota2CalendarSync:
def process_upcoming_matches(self, upcoming_matches: List[Match], existing_events: Dict[str, Any], 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]: dry_run: bool, update_results: bool, update_times: bool) -> Dict[str, int]:
"""Process upcoming matches and update calendar""" """Process upcoming matches and update calendar
Handles:
- Adding new matches
- Updating TBD placeholders with team names
- Updating in-progress scores
- Updating match times
Args:
upcoming_matches: List of upcoming Match objects
existing_events: Dictionary of existing calendar events
dry_run: If True, only simulate changes
update_results: Whether to update match scores
update_times: Whether to update match times
Returns:
Dictionary with counters for added, updated, skipped matches
"""
counters = { counters = {
'added': 0, 'added': 0,
'updated': 0, 'updated': 0,
@ -908,7 +1151,18 @@ class Dota2CalendarSync:
def process_completed_matches(self, completed_matches: List[Match], existing_events: Dict[str, Any], def process_completed_matches(self, completed_matches: List[Match], existing_events: Dict[str, Any],
dry_run: bool) -> Dict[str, int]: dry_run: bool) -> Dict[str, int]:
"""Process completed matches and update results""" """Process completed matches and update results
Updates calendar events with final scores and winners.
Args:
completed_matches: List of completed Match objects
existing_events: Dictionary of existing calendar events
dry_run: If True, only simulate changes
Returns:
Dictionary with counters for updated matches and errors
"""
counters = { counters = {
'updated': 0, 'updated': 0,
'errors': 0 'errors': 0
@ -947,7 +1201,21 @@ class Dota2CalendarSync:
def clean_duplicate_and_expired_events(self, existing_events: Dict[str, Any], def clean_duplicate_and_expired_events(self, existing_events: Dict[str, Any],
updated_tbd_events: set, dry_run: bool) -> int: updated_tbd_events: set, dry_run: bool) -> int:
"""Clean up duplicate and expired TBD events""" """Clean up duplicate and expired TBD events
Removes:
- TBD events that have ended
- Duplicate TBD vs TBD events at the same time
- TBD events superseded by confirmed matches
Args:
existing_events: Dictionary of existing calendar events
updated_tbd_events: Set of event IDs that were updated (to skip)
dry_run: If True, only simulate deletions
Returns:
Number of events deleted
"""
deleted_count = 0 deleted_count = 0
now = datetime.now(pytz.UTC) now = datetime.now(pytz.UTC)
@ -1045,7 +1313,21 @@ class Dota2CalendarSync:
return deleted_count return deleted_count
def sync_matches_to_calendar(self, dry_run=False, update_results=True, update_times=True, delete_old_tbd=True): 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""" """Main sync function with improved structure and error handling
Coordinates the entire sync process:
1. Fetches matches from Liquipedia
2. Gets existing events from Google Calendar
3. Processes upcoming and completed matches
4. Cleans up duplicates and expired TBD events
5. Displays summary
Args:
dry_run: If True, only simulate changes
update_results: Whether to update match scores/results
update_times: Whether to update changed match times
delete_old_tbd: Whether to delete expired TBD events
"""
logger.info("\n" + "="*50) logger.info("\n" + "="*50)
logger.info("Starting Dota 2 Calendar Sync v4.0") logger.info("Starting Dota 2 Calendar Sync v4.0")
logger.info("="*50 + "\n") logger.info("="*50 + "\n")
@ -1114,6 +1396,10 @@ class Dota2CalendarSync:
raise raise
def main(): def main():
"""Main entry point for the script
Parses command line arguments and runs the sync process.
"""
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description='Sync Dota 2 Tier 1 matches from Liquipedia to Google Calendar v4.0' description='Sync Dota 2 Tier 1 matches from Liquipedia to Google Calendar v4.0'
) )