cal/major-domo-v2
1
0
Fork 0
Code Issues 6 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

Merge pull request #3 from calcorum/enhancement/unpublish-scorecard

Browse Source 
Enhancement/unpublish scorecard
This commit is contained in:
Cal Corum 2025-10-21 14:47:08 -05:00 committed by GitHub
commit 225cbb8c3e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
20 changed files with 2127 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
bot.py
View File

@ -121,6 +121,7 @@ class SBABot(commands.Bot):
from commands.profile import setup_profile_commands
from commands.soak import setup_soak
from commands.injuries import setup_injuries
from commands.gameplay import setup_gameplay
# Define command packages to load
command_packages = [
@ -137,6 +138,7 @@ class SBABot(commands.Bot):
("profile", setup_profile_commands),
("soak", setup_soak),
("injuries", setup_injuries),
("gameplay", setup_gameplay),
]
total_successful = 0
@ -187,6 +189,11 @@ class SBABot(commands.Bot):
asyncio.create_task(self.voice_cleanup_service.start_monitoring(self))
self.logger.info("✅ Voice channel cleanup service started")
# Initialize live scorebug tracker
from tasks.live_scorebug_tracker import setup_scorebug_tracker
self.live_scorebug_tracker = setup_scorebug_tracker(self)
self.logger.info("✅ Live scorebug tracker started")
self.logger.info("✅ Background tasks initialized successfully")
except Exception as e:
@ -341,6 +348,13 @@ class SBABot(commands.Bot):
except Exception as e:
self.logger.error(f"Error stopping voice cleanup service: {e}")
if hasattr(self, 'live_scorebug_tracker'):
try:
self.live_scorebug_tracker.update_loop.cancel()
self.logger.info("Live scorebug tracker stopped")
except Exception as e:
self.logger.error(f"Error stopping live scorebug tracker: {e}")
# Call parent close method
await super().close()
self.logger.info("Bot shutdown complete")

259
commands/dice/rolls.py
View File

@ -17,6 +17,7 @@ from utils import team_utils
from utils.logging import get_contextual_logger
from utils.decorators import logged_command
from utils.team_utils import get_user_major_league_team
from utils.text_utils import split_text_for_fields
from views.embeds import EmbedColors, EmbedTemplate
@ -29,6 +30,127 @@ class DiceRoll:
rolls: list[int]
total: int
INFIELD_X_CHART = {
'si1': {
'rp': 'Runner on first: Line drive hits the runner! Runner on first is out. Batter goes to first with single '
'and all other runners hold.\nNo runner on first: batter singles, runners advance 1 base.',
'e1': 'Single and Error, batter to second, runners advance 2 bases.',
'e2': 'Single and Error, batter to third, all runners score.',
'no': 'Single, runners advance 1 base.'
},
'spd': {
'rp': 'No effect; proceed with speed check',
'e1': 'Single and Error, batter to second, runners advance 2 bases.',
'e2': 'Single and Error, batter to third, all runners score.',
'no': 'Speed check, safe range equals batter\'s running rating, SI* result if safe, gb C if out'
},
'po': {
'rp': 'The batters hits a popup. None of the fielders take charge on the play and the ball drops in the '
'infield for a single! All runners advance 1 base.',
'e1': 'The catcher drops a popup for an error. All runners advance 1 base.',
'e2': 'The catcher grabs a squib in front of the plate and throws it into right field. The batter goes to '
'second and all runners score.',
'no': 'The batter pops out to the catcher.'
},
'wp': {
'rp': 'Automatic wild pitch. Catcher has trouble finding it and all base runners advance 2 bases.',
'no': 'Automatic wild pitch, all runners advance 1 base and batter rolls AB again.'
},
'x': {
'rp': 'Runner(s) on base: pitcher trips during his delivery and the ball sails for automatic wild pitch, '
'runners advance 1 base and batter rolls AB again.',
'no': 'Wild pitch check (credited as a PB). If a passed ball occurs, batter rerolls AB. '
'If no passed ball occurs, the batter fouls out to the catcher.'
},
'fo': {
'rp': 'Batter swings and misses, but is awarded first base on a catcher interference call! Baserunners advance '
'only if forced.',
'e1': 'The catcher drops a foul popup for an error. Batter rolls AB again.',
'e2': 'The catcher drops a foul popup for an error. Batter rolls AB again.',
'no': 'Runner(s) on base: make a passed ball check. If no passed ball, batter pops out to the catcher. If a '
'passed ball occurs, batter roll his AB again.\nNo runners: batter pops out to the catcher'
},
'g1': {
'rp': 'Runner on first: runner on first breaks up the double play, but umpires call runner interference and '
'the batter is out on GIDP.\nNo runners: Batter grounds out.',
'e1': 'Error, batter to first, runners advance 1 base.',
'e2': 'Error, batter to second, runners advance 2 bases.',
'no': 'Consult Groundball Chart: `!gbA`'
},
'g2': {
'rp': 'Batter lines the ball off the pitcher to the fielder who makes the play to first for the out! Runners '
'advance only if forced.',
'e1': 'Error, batter to first, runners advance 1 base.',
'e2': 'Error, batter to second, runners advance 2 bases.',
'no': 'Consult Groundball Chart: `!gbB`'
},
'g3': {
'rp': 'Batter lines the ball off the mound and deflects to the fielder who makes the play to first for the '
'out! Runners advance 1 base.',
'e1': 'Error, batter to first, runners advance 1 base.',
'e2': 'Error, batter to second, runners advance 2 bases.',
'no': 'Consult Groundball Chart: `!gbC`'
},
}
OUTFIELD_X_CHART = {
'si2': {
'rp': 'Batter singles, baserunners advance 2 bases. As the batter rounds first, the fielder throws behind him '
'and catches him off the bag for an out!',
'e1': 'Single and error, batter to second, runners advance 2 bases.',
'e2': 'Single and error, batter to third, all runners score.',
'e3': 'Single and error, batter to third, all runners score',
'no': 'Single, all runners advance 2 bases.'
},
'do2': {
'rp': 'Batter doubles, runners advance 2 bases. The outfielder throws the ball to the shortstop who executes a '
'hidden ball trick! Runner on second is called out!',
'e1': 'Double and error, batter to third, all runners score.',
'e2': 'Double and error, batter to third, and all runners score.',
'e3': 'Double and error, batter and all runners score. Little league home run!',
'no': 'Double, all runners advance 2 bases.'
},
'do3': {
'rp': 'Runner(s) on base: batter doubles and runners advance three bases as the outfielders collide!\n'
'No runners: Batter doubles, but the play is appealed. The umps rule the batter missed first base so is '
'out on the appeal!',
'e1': 'Double and error, batter to third, all runners score.',
'e2': 'Double and error, batter and all runners score. Little league home run!',
'e3': 'Double and error, batter and all runners score. Little league home run!',
'no': 'Double, all runners score.'
},
'tr3': {
'rp': 'Batter hits a ball into the gap and the outfielders collide trying to make the play! The ball rolls to '
'the wall and the batter trots home with an inside-the-park home run!',
'e1': 'Triple and error, batter and all runners score. Little league home run!',
'e2': 'Triple and error, batter and all runners score. Little league home run!',
'e3': 'Triple and error, batter and all runners score. Little league home run!',
'no': 'Triple, all runners score.'
},
'f1': {
'rp': 'The outfielder races back and makes a diving catch and collides with the wall! In the time he takes to '
'recuperate, all baserunners tag-up and advance 2 bases.',
'e1': '1 base error, runners advance 1 base.',
'e2': '2 base error, runners advance 2 bases.',
'e3': '3 base error, batter to third, all runners score.',
'no': 'Flyball A'
},
'f2': {
'rp': 'The outfielder catches the flyball for an out. If there is a runner on third, he tags-up and scores. '
'The play is appealed and the umps rule that the runner left early and is out on the appeal!',
'e1': '1 base error, runners advance 1 base.',
'e2': '2 base error, runners advance 2 bases.',
'e3': '3 base error, batter to third, all runners score.',
'no': 'Flyball B'
},
'f3': {
'rp': 'The outfielder makes a running catch in the gap! The lead runner lost track of the ball and was '
'advancing - he cannot return in time and is doubled off by the outfielder.',
'e1': '1 base error, runners advance 1 base.',
'e2': '2 base error, runners advance 2 bases.',
'e3': '3 base error, batter to third, all runners score.',
'no': 'Flyball C'
}
}
class DiceRollCommands(commands.Cog):
"""Dice rolling command handlers for gameplay."""
@ -191,16 +313,17 @@ class DiceRollCommands(commands.Cog):
):
"""Roll Super Advanced fielding dice for a defensive position."""
await interaction.response.defer()
embed_color = await self._get_channel_embed_color(interaction)
# Get the position value from the choice
pos_value = position.value
# Roll the dice - 1d20 and 3d6
dice_notation = "1d20;3d6"
dice_notation = "1d20;3d6;1d100"
roll_results = self._parse_and_roll_multiple_dice(dice_notation)
# Create fielding embed
embed = self._create_fielding_embed(pos_value, roll_results, interaction.user)
embed = self._create_fielding_embed(pos_value, roll_results, interaction.user, embed_color)
await interaction.followup.send(embed=embed)
@commands.command(name="f", aliases=["fielding", "saf"])
@ -326,16 +449,23 @@ class DiceRollCommands(commands.Cog):
return position_map.get(pos)
def _create_fielding_embed(self, position: str, roll_results: list[DiceRoll], user) -> discord.Embed:
def _create_fielding_embed(
self,
position: str,
roll_results: list[DiceRoll],
user: discord.User | discord.Member,
embed_color: int = EmbedColors.PRIMARY
) -> discord.Embed:
"""Create an embed for fielding roll results."""
d20_result = roll_results[0].total
d6_total = roll_results[1].total
d6_rolls = roll_results[1].rolls
d100_result = roll_results[2].total
# Create base embed
embed = EmbedTemplate.create_base_embed(
title=f"SA Fielding roll for {user.display_name}",
color=EmbedColors.PRIMARY
color=embed_color
)
# Set user info
@ -364,21 +494,38 @@ class DiceRollCommands(commands.Cog):
inline=False
)
# Add error result
error_result = self._get_error_result(position, d6_total)
if error_result:
embed.add_field(
name="Error Result",
value=error_result,
inline=False
)
# Add rare play or error result
if d100_result == 1:
error_result = self._get_rare_play(position, d20_result)
base_field_name = "Rare Play Result"
else:
# Add error result
error_result = self._get_error_result(position, d6_total)
base_field_name = "Error Result"
# # Add help commands
# embed.add_field(
# name="Help Commands",
# value="Run !<result> for full chart readout (e.g. !g1 or !do3)",
# inline=False
# )
if error_result:
# Split text if it exceeds Discord's field limit
result_chunks = split_text_for_fields(error_result, max_length=1024)
# Add each chunk as a separate field
for i, chunk in enumerate(result_chunks):
field_name = base_field_name
# Add part indicator if multiple chunks
if len(result_chunks) > 1:
field_name += f" (Part {i+1}/{len(result_chunks)})"
embed.add_field(
name=field_name,
value=chunk,
inline=False
)
# Add help commands
embed.add_field(
name="Help Commands",
value="Run /charts for full chart readout",
inline=False
)
# # Add references
# embed.add_field(
@ -564,6 +711,26 @@ class DiceRollCommands(commands.Cog):
20: '--------G1---------'
}
return pitcher_ranges.get(d20_roll, 'Unknown')
def _get_rare_play(self, position: str, d20_total: int) -> str:
"""Get the rare play result for a position and d20 total"""
starter = 'Rare play! Take the range result from above and consult the chart below.\n\n'
if position == 'P':
return starter + self._get_pitcher_rare_play(d20_total)
elif position == '1B':
return starter + self._get_infield_rare_play(d20_total)
elif position == '2B':
return starter + self._get_infield_rare_play(d20_total)
elif position == '3B':
return starter + self._get_infield_rare_play(d20_total)
elif position == 'SS':
return starter + self._get_infield_rare_play(d20_total)
elif position in ['LF', 'RF']:
return starter + self._get_outfield_rare_play(d20_total)
elif position == 'CF':
return starter + self._get_outfield_rare_play(d20_total)
raise ValueError(f'Unknown position: {position}')
def _get_error_result(self, position: str, d6_total: int) -> str:
"""Get the error result for a position and 3d6 total."""
@ -723,22 +890,22 @@ class DiceRollCommands(commands.Cog):
def _get_catcher_error(self, d6_total: int) -> str:
"""Get Catcher error result based on 3d6 total."""
errors = {
18: 'Passed ball for sb2 -> sb12, sb16 -> sb26\nNo error for sb14',
17: 'Passed ball for sb3 -> sb12, sb17 -> sb26\nNo error for sb1, sb13 -> sb15',
16: 'Passed ball for sb4 -> sb12, sb18 -> sb26',
15: 'Passed ball for sb5 -> sb12, sb19 -> sb26',
14: 'Passed ball for sb6 -> sb12, sb20 -> sb26',
13: 'Passed ball for sb7 -> sb12, sb21 -> sb26',
12: 'Passed ball for sb8 -> sb12, sb22 -> sb26',
11: 'Passed ball for sb9 -> sb12, sb23 -> sb26',
10: 'Passed ball for sb10 -> sb12, sb24 -> sb26',
9: 'Passed ball for sb11, sb12, sb25, sb26',
18: '2-base error for e4 -> 16\n1-base error for e2, e3',
17: '1-base error for e1, e2, e4, e5, e12 -> e14, e16',
16: '1-base error for e3 -> e5, e7, e12 -> e14, e16',
15: '1-base error for e7, e8, e12, e13, e15',
14: '1-base error for e6',
13: '1-base error for e9',
12: '1-base error for e10, e14',
11: '1-base error for e11, e15',
10: 'No error',
9: 'No error',
8: 'No error',
7: 'No error',
6: 'No error',
7: '1-base error for e16',
6: '1-base error for e8, e12, e13',
5: 'No error',
4: 'Passed ball for sb1 -> sb12, sb15 -> sb26\nNo error for sb13, sb14',
3: 'Passed ball for sb1 -> sb26'
4: '1-base error for e5, e13',
3: '2-base error for e12 -> e16\n1-base error for e2, e3, e7, e11'
}
return errors.get(d6_total, 'No error')
@ -764,6 +931,34 @@ class DiceRollCommands(commands.Cog):
}
return errors.get(d6_total, 'No error')
def _get_pitcher_rare_play(self, d20_total: int) -> str:
return (
f'**G3**: {INFIELD_X_CHART["g3"]["rp"]}\n'
f'**G2**: {INFIELD_X_CHART["g2"]["rp"]}\n'
f'**G1**: {INFIELD_X_CHART["g1"]["rp"]}\n'
f'**SI1**: {INFIELD_X_CHART["si1"]["rp"]}\n'
)
def _get_infield_rare_play(self, d20_total: int) -> str:
return (
f'**G3**: {INFIELD_X_CHART["g3"]["rp"]}\n'
f'**G2**: {INFIELD_X_CHART["g2"]["rp"]}\n'
f'**G1**: {INFIELD_X_CHART["g1"]["rp"]}\n'
f'**SI1**: {INFIELD_X_CHART["si1"]["rp"]}\n'
f'**SI2**: {OUTFIELD_X_CHART["si2"]["rp"]}\n'
)
def _get_outfield_rare_play(self, d20_total: int) -> str:
return (
f'**F1**: {OUTFIELD_X_CHART["f1"]["rp"]}\n'
f'**F2**: {OUTFIELD_X_CHART["f2"]["rp"]}\n'
f'**F3**: {OUTFIELD_X_CHART["f3"]["rp"]}\n'
f'**SI2**: {OUTFIELD_X_CHART["si2"]["rp"]}\n'
f'**DO2**: {OUTFIELD_X_CHART["do2"]["rp"]}\n'
f'**DO3**: {OUTFIELD_X_CHART["do3"]["rp"]}\n'
f'**TR3**: {OUTFIELD_X_CHART["tr3"]["rp"]}\n'
)
def _parse_and_roll_multiple_dice(self, dice_notation: str) -> list[DiceRoll]:
"""Parse dice notation (supports multiple rolls) and return roll results."""
# Split by semicolon for multiple rolls

317
commands/gameplay/CLAUDE.md Normal file
View File

@ -0,0 +1,317 @@
# Gameplay Commands
This directory contains Discord slash commands for live game tracking and scorecard management during gameplay.
## Files
### `scorebug.py`
- **Commands**:
- `/publish-scorecard <url>` - Link a Google Sheets scorecard to a channel for live tracking
- `/scorebug [full_length]` - Display the current scorebug from the published scorecard
- **Description**: Main command implementation for scorebug display and management
- **Service Dependencies**:
- `ScorebugService` - Reading live game data from Google Sheets
- `team_service.get_team()` - Team information lookup
- **Tracker Dependencies**:
- `ScorecardTracker` - JSON-based persistent storage of scorecard-channel mappings
### `scorecard_tracker.py`
- **Class**: `ScorecardTracker`
- **Description**: JSON-based persistent tracking of published scorecards
- **Features**:
- Maps Discord text channels to Google Sheets URLs
- Persistent storage across bot restarts
- Automatic stale entry cleanup
- Timestamp tracking for monitoring
### `__init__.py`
- **Function**: `setup_gameplay(bot)`
- **Description**: Package initialization with resilient cog loading
- **Integration**: Follows established bot architecture patterns
## Background Integration
### Live Scorebug Tracker Task
**File**: `tasks/live_scorebug_tracker.py`
**Schedule**: Every 3 minutes
**Operations**:
1. **Update `#live-sba-scores` Channel**
- Reads all published scorecards from tracker
- Generates compact scorebug embeds for active games
- Clears old messages and posts fresh scorebugs
- Filters out final games (only shows active/in-progress)
2. **Update Voice Channel Descriptions**
- For each active scorecard, checks for associated voice channel
- Updates voice channel topic with live score: `"BOS 4 @ 3 NYY"`
- Adds "FINAL" suffix when game completes: `"BOS 5 @ 3 NYY - FINAL"`
- Gracefully handles missing or deleted voice channels
## Key Features
### `/publish-scorecard <url>`
**URL/Key Support**:
- Full URL: `https://docs.google.com/spreadsheets/d/[KEY]/edit...`
- Just the key: `[SHEET_KEY]`
**Validation**:
- Checks scorecard accessibility (public read permissions)
- Verifies scorecard has required 'Scorebug' tab
- Tests data reading to ensure valid scorecard structure
**Storage**:
- Saves mapping in `data/scorecards.json`
- Persists across bot restarts
- Associates scorecard with text channel ID
**User Feedback**:
- Confirmation message with sheet title
- Usage instructions for `/scorebug` command
- Clear error messages for access issues
### `/scorebug [full_length]`
**Display Modes**:
- `full_length=True` (default): Complete scorebug with runners, matchups, and summary
- `full_length=False`: Compact view with just score and status
**Data Display**:
- Game header and inning information
- Score formatted in table
- Current game status (inning/half)
- Runners on base with positions
- Current matchups (optional)
- Game summary (optional)
- Team colors and thumbnails
**Error Handling**:
- Clear message if no scorecard published in channel
- Helpful errors for access or read failures
- Graceful handling of missing team data
### Live Score Updates (Background Task)
**Channel Updates**:
- Clears `#live-sba-scores` channel before each update
- Posts up to 10 scorebugs per message (Discord limit)
- Splits into multiple messages if needed
- Shows only active games (filters out finals)
**Voice Channel Integration**:
- Looks up voice channel associated with scorecard's text channel
- Updates voice channel `topic` with formatted score
- Format: `"{AWAY_ABBREV} {AWAY_SCORE} @ {HOME_SCORE} {HOME_ABBREV}"`
- Adds "- FINAL" when game completes
- Rate limits to avoid Discord API issues
## Architecture
### Service Layer Integration
**ScorebugService** (`services/scorebug_service.py`):
- Extends `SheetsService` for Google Sheets access
- Returns `ScorebugData` objects with parsed game information
- Supports both URL and key-based scorecard access
- Reads from 'Scorebug' tab (B2:S20) for game state
- Reads team IDs from 'Setup' tab (B5:B6)
**ScorebugData Fields**:
```python
{
'away_team_id': int,
'home_team_id': int,
'header': str, # Game header with inning info
'away_score': int,
'home_score': int,
'which_half': str, # Top/Bottom inning indicator
'is_final': bool,
'runners': list, # Runner info [position, name] pairs
'matchups': list, # Current batter/pitcher matchups
'summary': list # Game summary data
}
```
### Tracker Integration
**ScorecardTracker** stores:
```json
{
"scorecards": {
"123456789": {
"text_channel_id": "123456789",
"sheet_url": "https://docs.google.com/...",
"published_at": "2025-01-15T10:30:00",
"last_updated": "2025-01-15T10:35:00",
"publisher_id": "111222333"
}
}
}
```
**Voice Channel Association**:
- Voice tracker updated to store `text_channel_id` when voice channels created
- New method: `get_voice_channel_for_text_channel(text_channel_id)`
- Enables background task to update voice channel descriptions
### Command Flow
**Publishing Flow**:
1. User runs `/publish-scorecard <url>`
2. Bot validates access to Google Sheet
3. Bot verifies Scorebug tab exists
4. Bot reads sample data to ensure valid structure
5. Tracker stores text_channel_id → sheet_url mapping
6. User receives confirmation message
**Display Flow**:
1. User runs `/scorebug` in channel
2. Bot looks up scorecard URL from tracker
3. Bot reads current scorebug data from sheet
4. Bot fetches team information from API
5. Bot creates rich embed with game state
6. Bot updates tracker timestamp
**Background Update Flow**:
1. Task runs every 3 minutes
2. Reads all published scorecards from tracker
3. For each scorecard:
- Reads current scorebug data
- Checks if game is active (not final)
- Creates compact embed for live channel
- Checks for associated voice channel
- Updates voice channel description if found
4. Posts all active scorebugs to `#live-sba-scores`
5. Clears channel if no active games
## Configuration
### Channel Requirements
- **`#live-sba-scores`** - Live scorebug display channel (auto-updated every 3 minutes)
### Data Storage
- **`data/scorecards.json`** - Published scorecard mappings
- **`data/voice_channels.json`** - Voice channel tracking (includes text_channel_id)
### Google Sheets Requirements
Scorecards must have:
- **Scorebug tab**: Live game data (B2:S20)
- **Setup tab**: Team IDs (B5:B6)
- **Public read access**: "Anyone with the link can view"
## Error Handling
### Common Scenarios
- **Sheet not accessible**: Clear message about public permissions
- **Missing Scorebug tab**: Error indicating invalid scorecard structure
- **No scorecard published**: Helpful message to use `/publish-scorecard`
- **Sheet read failures**: Graceful degradation with retry suggestions
- **Voice channel deleted**: Silent skip (no errors to users)
- **Missing permissions**: Clear permission error messages
### Service Dependencies
- **Graceful degradation**: Commands work without background task
- **Rate limiting**: 1-second delay between scorecard reads
- **API failures**: Comprehensive error handling for external service calls
- **Discord errors**: Specific handling for Forbidden, NotFound, etc.
## Voice Channel Enhancement
### Text Channel Association
When voice channels are created via `/voice-channel`:
- Text channel ID stored in voice channel tracking data
- Enables scorebug → voice channel lookup
- Persistent across bot restarts
### Description Update Format
**Active Game**:
```
BOS 4 @ 3 NYY
```
**Final Game**:
```
BOS 5 @ 3 NYY - FINAL
```
**Implementation**:
- Uses voice channel `topic` field (description)
- Updates every 3 minutes with live scores
- Automatic cleanup when game ends
- No manual user interaction required
## Integration Points
### Bot Integration
- **Package Loading**: Integrated into `bot.py` command package loading sequence
- **Background Tasks**: Live scorebug tracker started in `_setup_background_tasks()`
- **Shutdown Handling**: Tracker stopped in `bot.close()`
### Service Layer
- **ScorebugService**: Google Sheets data extraction
- **TeamService**: Team information and logo lookups
- **ScorecardTracker**: Persistent scorecard-channel mapping
### Discord Integration
- **Application Commands**: Modern slash command interface
- **Embed Templates**: Consistent styling using `EmbedTemplate`
- **Error Handling**: Integration with global application command error handler
- **Voice Channels**: Bi-directional integration with voice channel system
## Usage Examples
### Publishing a Scorecard
```
/publish-scorecard https://docs.google.com/spreadsheets/d/ABC123/edit
```
**Result**: Scorecard linked to current channel for live tracking
### Using Just Sheet Key
```
/publish-scorecard ABC123DEF456
```
**Result**: Same functionality with cleaner input
### Displaying Scorebug
```
/scorebug
```
**Result**: Full scorebug display with all details
### Compact Scorebug
```
/scorebug full_length:False
```
**Result**: Just score and status, no runners/matchups/summary
## Monitoring and Logs
### Structured Logging
```python
self.logger.info(f"Published scorecard to channel {text_channel_id}: {sheet_url}")
self.logger.debug(f"Updated voice channel {voice_channel.name} description to: {description}")
self.logger.warning(f"Could not read scorecard {sheet_url}: {e}")
```
### Performance Tracking
- Background task execution timing
- Google Sheets read latency
- Voice channel update success rates
- Scorecard access failure rates
## Future Enhancements
### Potential Features
- **Scorecard rotation**: Multiple scorecard support per channel
- **Custom refresh intervals**: User-configurable update frequency
- **Notification system**: Alerts for game events (runs, innings, etc.)
- **Statistics tracking**: Historical scorebug access patterns
- **Mobile optimization**: Compact embeds for mobile viewing
### Configuration Options
- **Per-channel settings**: Different update intervals per channel
- **Role permissions**: Restrict scorecard publishing to specific roles
- **Format customization**: User-selectable scorebug styles
- **Alert thresholds**: Configurable notification triggers
---
**Last Updated**: January 2025
**Architecture**: Modern async Discord.py with Google Sheets integration
**Dependencies**: discord.py, pygsheets, ScorebugService, ScorecardTracker, VoiceChannelTracker

50
commands/gameplay/__init__.py Normal file
View File

@ -0,0 +1,50 @@
"""
Gameplay Commands Package
This package contains commands for live game tracking and scorecard management.
"""
import logging
from discord.ext import commands
from .scorebug import ScorebugCommands
logger = logging.getLogger(__name__)
async def setup_gameplay(bot: commands.Bot):
"""
Setup all gameplay command modules.
Returns:
tuple: (successful_count, failed_count, failed_modules)
"""
# Define all gameplay command cogs to load
gameplay_cogs = [
("ScorebugCommands", ScorebugCommands),
]
successful = 0
failed = 0
failed_modules = []
for cog_name, cog_class in gameplay_cogs:
try:
await bot.add_cog(cog_class(bot))
logger.info(f"✅ Loaded {cog_name}")
successful += 1
except Exception as e:
logger.error(f"❌ Failed to load {cog_name}: {e}", exc_info=True)
failed += 1
failed_modules.append(cog_name)
# Log summary
if failed == 0:
logger.info(f"🎉 All {successful} gameplay command modules loaded successfully")
else:
logger.warning(f"⚠️ Gameplay commands loaded with issues: {successful} successful, {failed} failed")
return successful, failed, failed_modules
# Export the setup function for easy importing
__all__ = ['setup_gameplay', 'ScorebugCommands']

358
commands/gameplay/scorebug.py Normal file
View File

@ -0,0 +1,358 @@
"""
Scorebug Commands
Implements commands for publishing and displaying live game scorebugs from Google Sheets scorecards.
"""
import discord
from discord.ext import commands
from discord import app_commands
from services.scorebug_service import ScorebugService
from services.team_service import team_service
from utils.logging import get_contextual_logger
from utils.decorators import logged_command
from views.embeds import EmbedTemplate, EmbedColors
from exceptions import SheetsException
from .scorecard_tracker import ScorecardTracker
class ScorebugCommands(commands.Cog):
"""Scorebug command handlers for live game tracking."""
def __init__(self, bot: commands.Bot):
self.bot = bot
self.logger = get_contextual_logger(f'{__name__}.ScorebugCommands')
self.scorebug_service = ScorebugService()
self.scorecard_tracker = ScorecardTracker()
self.logger.info("ScorebugCommands cog initialized")
@app_commands.command(
name="publish-scorecard",
description="Publish a Google Sheets scorecard to this channel for live tracking"
)
@app_commands.describe(
url="Full URL to the Google Sheets scorecard or just the sheet key"
)
@logged_command("/publish-scorecard")
async def publish_scorecard(
self,
interaction: discord.Interaction,
url: str
):
"""
Link a Google Sheets scorecard to the current channel for live scorebug tracking.
The scorecard will be monitored for live score updates which will be displayed
in the live scores channel and optionally in associated voice channels.
"""
await interaction.response.defer()
try:
# Validate access to the scorecard
await interaction.edit_original_response(
content="📋 Accessing scorecard..."
)
# Try to open the scorecard to validate it
scorecard = await self.scorebug_service.open_scorecard(url)
# Verify it has a Scorebug tab
try:
scorebug_data = await self.scorebug_service.read_scorebug_data(url, full_length=False)
except SheetsException:
embed = EmbedTemplate.error(
title="Invalid Scorecard",
description=(
"This doesn't appear to be a valid scorecard.\n\n"
"Make sure the sheet has a 'Scorebug' tab and is properly set up."
)
)
await interaction.edit_original_response(content=None, embed=embed)
return
# Get team data for display
away_team = None
home_team = None
if scorebug_data.away_team_id:
away_team = await team_service.get_team(scorebug_data.away_team_id)
if scorebug_data.home_team_id:
home_team = await team_service.get_team(scorebug_data.home_team_id)
# Format scorecard link
away_abbrev = away_team.abbrev if away_team else "AWAY"
home_abbrev = home_team.abbrev if home_team else "HOME"
scorecard_link = f"[{away_abbrev} @ {home_abbrev}]({url})"
# Store the scorecard in the tracker
self.scorecard_tracker.publish_scorecard(
text_channel_id=interaction.channel_id, # type: ignore
sheet_url=url,
publisher_id=interaction.user.id
)
# Create success embed
embed = EmbedTemplate.success(
title="Scorecard Published",
description=(
f"Your scorecard has been published to {interaction.channel.mention}\n\n" # type: ignore
f"**Sheet:** {scorecard.title}\n"
f"**Status:** Live tracking enabled\n"
f"**Scorecard:** {scorecard_link}\n\n"
f"Anyone can now run `/scorebug` in this channel to see the current score.\n"
f"The scorebug will also update in the live scores channel every 3 minutes."
)
)
embed.add_field(
name="Commands",
value=(
"`/scorebug` - Display full scorebug with details\n"
"`/scorebug full_length:False` - Display compact scorebug"
),
inline=False
)
await interaction.edit_original_response(content=None, embed=embed)
except SheetsException as e:
embed = EmbedTemplate.error(
title="Cannot Access Scorecard",
description=(
f"❌ {str(e)}\n\n"
f"**Common issues:**\n"
f"• Sheet is not publicly accessible\n"
f"• Invalid sheet URL or key\n"
f"• Sheet doesn't exist\n\n"
f"Make sure your sheet is shared with 'Anyone with the link can view'."
)
)
await interaction.edit_original_response(content=None, embed=embed)
except Exception as e:
self.logger.error(f"Error publishing scorecard: {e}", exc_info=True)
embed = EmbedTemplate.error(
title="Publication Failed",
description=(
"❌ An unexpected error occurred while publishing the scorecard.\n\n"
"Please try again or contact support if the issue persists."
)
)
await interaction.edit_original_response(content=None, embed=embed)
@app_commands.command(
name="scorebug",
description="Display the scorebug for the game in this channel"
)
@app_commands.describe(
full_length="Include full game details (defaults to True)"
)
@logged_command("/scorebug")
async def scorebug(
self,
interaction: discord.Interaction,
full_length: bool = True
):
"""
Display the current scorebug from the scorecard published in this channel.
"""
await interaction.response.defer()
# Check if a scorecard is published in this channel
sheet_url = self.scorecard_tracker.get_scorecard(interaction.channel_id) # type: ignore
if not sheet_url:
embed = EmbedTemplate.error(
title="No Scorecard Published",
description=(
"❌ No scorecard has been published in this channel.\n\n"
"Use `/publish-scorecard <url>` to publish a scorecard first."
)
)
await interaction.followup.send(embed=embed, ephemeral=True)
return
try:
# Read scorebug data
await interaction.edit_original_response(
content="📊 Reading scorebug..."
)
scorebug_data = await self.scorebug_service.read_scorebug_data(
sheet_url,
full_length=full_length
)
# Get team data
away_team = None
home_team = None
if scorebug_data.away_team_id:
away_team = await team_service.get_team(scorebug_data.away_team_id)
if scorebug_data.home_team_id:
home_team = await team_service.get_team(scorebug_data.home_team_id)
# Create scorebug embed
embed = await self._create_scorebug_embed(
scorebug_data,
away_team,
home_team,
full_length
)
await interaction.edit_original_response(content=None, embed=embed)
# Update timestamp in tracker
self.scorecard_tracker.update_timestamp(interaction.channel_id) # type: ignore
except SheetsException as e:
embed = EmbedTemplate.error(
title="Cannot Read Scorebug",
description=(
f"❌ {str(e)}\n\n"
f"The scorecard may have been deleted or the sheet structure changed."
)
)
await interaction.edit_original_response(content=None, embed=embed)
except Exception as e:
self.logger.error(f"Error displaying scorebug: {e}", exc_info=True)
embed = EmbedTemplate.error(
title="Display Failed",
description=(
"❌ An error occurred while reading the scorebug.\n\n"
"Please try again or republish the scorecard."
)
)
await interaction.edit_original_response(content=None, embed=embed)
async def _create_scorebug_embed(
self,
scorebug_data,
away_team,
home_team,
full_length: bool
) -> discord.Embed:
"""
Create a rich embed from scorebug data.
Args:
scorebug_data: ScorebugData object
away_team: Away team object (optional)
home_team: Home team object (optional)
full_length: Include full details
Returns:
Discord embed with scorebug information
"""
# Determine winning team for embed color
if scorebug_data.away_score > scorebug_data.home_score and away_team:
embed_color = away_team.get_color_int()
thumbnail_url = away_team.thumbnail if away_team.thumbnail else None
elif scorebug_data.home_score > scorebug_data.away_score and home_team:
embed_color = home_team.get_color_int()
thumbnail_url = home_team.thumbnail if home_team.thumbnail else None
else:
embed_color = EmbedColors.INFO
thumbnail_url = None
# Create embed with header as title
embed = discord.Embed(
title=scorebug_data.header,
color=embed_color
)
if thumbnail_url:
embed.set_thumbnail(url=thumbnail_url)
# Add score information
away_abbrev = away_team.abbrev if away_team else "AWAY"
home_abbrev = home_team.abbrev if home_team else "HOME"
score_text = (
f"```\n"
f"{away_abbrev:<6} {scorebug_data.away_score:>3}\n"
f"{home_abbrev:<6} {scorebug_data.home_score:>3}\n"
f"```"
)
embed.add_field(
name="Score",
value=score_text,
inline=True
)
# Add game state
if not scorebug_data.is_final:
embed.add_field(
name="Status",
value=f"**{scorebug_data.which_half}**",
inline=True
)
# Add runners on base if present
if scorebug_data.runners and any(scorebug_data.runners):
runners_text = self._format_runners(scorebug_data.runners)
if runners_text:
embed.add_field(
name="Runners",
value=runners_text,
inline=False
)
# Add matchups if full length
if full_length and scorebug_data.matchups and any(scorebug_data.matchups):
matchups_text = self._format_matchups(scorebug_data.matchups)
if matchups_text:
embed.add_field(
name="Matchups",
value=matchups_text,
inline=False
)
# Add summary if full length
if full_length and scorebug_data.summary and any(scorebug_data.summary):
summary_text = self._format_summary(scorebug_data.summary)
if summary_text:
embed.add_field(
name="Summary",
value=summary_text,
inline=False
)
return embed
def _format_runners(self, runners) -> str:
"""Format runners on base for display."""
# runners is a list of [runner_name, runner_position] pairs
runner_lines = []
for runner_data in runners:
if runner_data and len(runner_data) >= 2 and runner_data[0]:
runner_lines.append(f"**{runner_data[1]}:** {runner_data[0]}")
return "\n".join(runner_lines) if runner_lines else ""
def _format_matchups(self, matchups) -> str:
"""Format current matchups for display."""
# matchups is a list of [batter, pitcher] pairs
matchup_lines = []
for matchup_data in matchups:
if matchup_data and len(matchup_data) >= 2 and matchup_data[0]:
matchup_lines.append(f"{matchup_data[0]} vs {matchup_data[1]}")
return "\n".join(matchup_lines) if matchup_lines else ""
def _format_summary(self, summary) -> str:
"""Format game summary for display."""
# summary is a list of summary line pairs
summary_lines = []
for summary_data in summary:
if summary_data and len(summary_data) >= 1 and summary_data[0]:
# Join both columns if present
line = " - ".join([str(x) for x in summary_data if x])
summary_lines.append(line)
return "\n".join(summary_lines) if summary_lines else ""
async def setup(bot: commands.Bot):
"""Load the scorebug commands cog."""
await bot.add_cog(ScorebugCommands(bot))

177
commands/gameplay/scorecard_tracker.py Normal file
View File

@ -0,0 +1,177 @@
"""
Scorecard Tracker
Provides persistent tracking of published scorecards per Discord text channel using JSON file storage.
"""
import json
import logging
from datetime import datetime, UTC
from pathlib import Path
from typing import Dict, List, Optional, Tuple
logger = logging.getLogger(f'{__name__}.ScorecardTracker')
class ScorecardTracker:
"""
Tracks published Google Sheets scorecards linked to Discord text channels.
Features:
- Persistent storage across bot restarts
- Channel-to-scorecard URL mapping
- Automatic stale entry cleanup
- Timestamp tracking for monitoring
"""
def __init__(self, data_file: str = "data/scorecards.json"):
"""
Initialize the scorecard tracker.
Args:
data_file: Path to the JSON data file
"""
self.data_file = Path(data_file)
self.data_file.parent.mkdir(exist_ok=True)
self._data: Dict[str, any] = {}
self.load_data()
def load_data(self) -> None:
"""Load scorecard data from JSON file."""
try:
if self.data_file.exists():
with open(self.data_file, 'r') as f:
self._data = json.load(f)
logger.debug(f"Loaded {len(self._data.get('scorecards', {}))} tracked scorecards")
else:
self._data = {"scorecards": {}}
logger.info("No existing scorecard data found, starting fresh")
except Exception as e:
logger.error(f"Failed to load scorecard data: {e}")
self._data = {"scorecards": {}}
def save_data(self) -> None:
"""Save scorecard data to JSON file."""
try:
with open(self.data_file, 'w') as f:
json.dump(self._data, f, indent=2, default=str)
logger.debug("Scorecard data saved successfully")
except Exception as e:
logger.error(f"Failed to save scorecard data: {e}")
def publish_scorecard(
self,
text_channel_id: int,
sheet_url: str,
publisher_id: int
) -> None:
"""
Link a scorecard to a text channel.
Args:
text_channel_id: Discord text channel ID
sheet_url: Google Sheets URL or key
publisher_id: Discord user ID who published the scorecard
"""
self._data.setdefault("scorecards", {})[str(text_channel_id)] = {
"text_channel_id": str(text_channel_id),
"sheet_url": sheet_url,
"published_at": datetime.now(UTC).isoformat(),
"last_updated": datetime.now(UTC).isoformat(),
"publisher_id": str(publisher_id)
}
self.save_data()
logger.info(f"Published scorecard to channel {text_channel_id}: {sheet_url}")
def unpublish_scorecard(self, text_channel_id: int) -> bool:
"""
Remove scorecard from a text channel.
Args:
text_channel_id: Discord text channel ID
Returns:
True if scorecard was removed, False if not found
"""
scorecards = self._data.get("scorecards", {})
channel_key = str(text_channel_id)
if channel_key in scorecards:
del scorecards[channel_key]
self.save_data()
logger.info(f"Unpublished scorecard from channel {text_channel_id}")
return True
return False
def get_scorecard(self, text_channel_id: int) -> Optional[str]:
"""
Get scorecard URL for a text channel.
Args:
text_channel_id: Discord text channel ID
Returns:
Sheet URL if published, None otherwise
"""
scorecards = self._data.get("scorecards", {})
scorecard_data = scorecards.get(str(text_channel_id))
return scorecard_data["sheet_url"] if scorecard_data else None
def get_all_scorecards(self) -> List[Tuple[int, str]]:
"""
Get all published scorecards.
Returns:
List of (text_channel_id, sheet_url) tuples
"""
scorecards = self._data.get("scorecards", {})
return [
(int(channel_id), data["sheet_url"])
for channel_id, data in scorecards.items()
]
def update_timestamp(self, text_channel_id: int) -> None:
"""
Update the last_updated timestamp for a scorecard.
Args:
text_channel_id: Discord text channel ID
"""
scorecards = self._data.get("scorecards", {})
channel_key = str(text_channel_id)
if channel_key in scorecards:
scorecards[channel_key]["last_updated"] = datetime.now(UTC).isoformat()
self.save_data()
def cleanup_stale_entries(self, valid_channel_ids: List[int]) -> int:
"""
Remove tracking entries for text channels that no longer exist.
Args:
valid_channel_ids: List of channel IDs that still exist in Discord
Returns:
Number of stale entries removed
"""
scorecards = self._data.get("scorecards", {})
stale_entries = []
for channel_id_str in scorecards.keys():
try:
channel_id = int(channel_id_str)
if channel_id not in valid_channel_ids:
stale_entries.append(channel_id_str)
except (ValueError, TypeError):
logger.warning(f"Invalid channel ID in scorecard data: {channel_id_str}")
stale_entries.append(channel_id_str)
# Remove stale entries
for channel_id_str in stale_entries:
del scorecards[channel_id_str]
logger.info(f"Removed stale scorecard entry for channel ID: {channel_id_str}")
if stale_entries:
self.save_data()
return len(stale_entries)

31
commands/injuries/management.py
View File

@ -406,13 +406,27 @@ class InjuryGroup(app_commands.Group):
# Check if player already has an active injury
existing_injury = await injury_service.get_active_injury(player.id, current.season)
# Data consistency check: If injury exists but il_return is None, it's stale data
if existing_injury:
embed = EmbedTemplate.error(
title="Already Injured",
description=f"Hm. It looks like {player.name} is already hurt."
)
await interaction.followup.send(embed=embed, ephemeral=True)
return
if not player.il_return:
# Stale injury record - clear it automatically
self.logger.warning(
f"Found stale injury record for {player.name} (injury {existing_injury.id}): "
f"is_active=True but il_return=None. Auto-clearing stale record."
)
await injury_service.clear_injury(existing_injury.id)
# Notify user but allow them to proceed
self.logger.info(f"Cleared stale injury {existing_injury.id} for player {player.id}")
else:
# Valid active injury - player is actually injured
embed = EmbedTemplate.error(
title="Already Injured",
description=f"Hm. It looks like {player.name} is already hurt (returns {player.il_return})."
)
await interaction.followup.send(embed=embed, ephemeral=True)
return
# Calculate return date
out_weeks = math.floor(injury_games / 4)
@ -610,7 +624,10 @@ class InjuryGroup(app_commands.Group):
inline=True
)
if player.team.roster_type() != RosterType.MAJOR_LEAGUE:
# Initialize responder_team to None for major league teams
if player.team.roster_type() == RosterType.MAJOR_LEAGUE:
responder_team = player.team
else:
responder_team = await team_utils.get_user_major_league_team(interaction.user.id)
# Create callback for confirmation

42
commands/league/standings.py
View File

@ -9,6 +9,7 @@ import discord
from discord.ext import commands
from config import get_config
from models.team import Team
from services.standings_service import standings_service
from utils.logging import get_contextual_logger
from utils.decorators import logged_command
@ -101,13 +102,15 @@ class StandingsCommands(commands.Cog):
embed = await self._create_division_embed(div_name, teams, season)
embeds.append(embed)
# Send first embed, then follow up with others
if embeds:
await interaction.followup.send(embed=embeds[0])
await interaction.followup.send(embeds=embeds)
# # Send first embed, then follow up with others
# if embeds:
# await interaction.followup.send(embed=embeds[0])
# Send additional embeds as follow-ups
for embed in embeds[1:]:
await interaction.followup.send(embed=embed)
# # Send additional embeds as follow-ups
# for embed in embeds[1:]:
# await interaction.followup.send(embed=embed)
async def _show_division_standings(self, interaction: discord.Interaction, season: int, division: str):
"""Show standings for a specific division."""
@ -147,28 +150,37 @@ class StandingsCommands(commands.Cog):
async def _create_division_embed(self, division_name: str, teams, season: int) -> discord.Embed:
"""Create an embed for a division's standings."""
embed = EmbedTemplate.create_base_embed(
title=f"🏆 {division_name} Division - Season {season}",
title=f"{division_name} Division",
description=f'Season {season}',
color=EmbedColors.PRIMARY
)
lead_team = None
# Create standings table
standings_lines = []
standing_text = f'```\n# Team W-L Win% RD GB St'
for i, team in enumerate(teams, 1):
if lead_team is None:
lead_team = team.team
if lead_team.color is not None:
embed.color = int(lead_team.color, 16)
embed.description = f'Leader: {lead_team.lname}'
# Format team line
team_line = (
f"{i}. **{team.team.abbrev}** {team.wins}-{team.losses} "
f"({team.winning_percentage:.3f})"
)
standing_text += f'\n{i}.{team.team.abbrev: >4} {team.wins: >2}-{team.losses: <2} {team.winning_percentage:.3f} {team.run_diff: >4}'
# Add games behind if not first place
if team.div_gb is not None and team.div_gb > 0:
team_line += f" *{team.div_gb:.1f} GB*"
standing_text += f' {team.div_gb:4.1f}'
else:
standing_text += f' '
standings_lines.append(team_line)
standing_text += f' {team.streak_wl.upper()}{team.streak_num}'
standing_text += '\n```'
embed.add_field(
name="Standings",
value="\n".join(standings_lines),
value=standing_text,
inline=False
)

2
commands/players/info.py
View File

@ -161,7 +161,7 @@ class PlayerInfoCommands(commands.Cog):
season=search_season,
batting_stats=batting_stats,
pitching_stats=pitching_stats,
user_id=interaction.user.id
user_id=None # setting to None so any GM can toggle the stats views
)
# Get initial embed with stats hidden

29
commands/voice/CLAUDE.md
View File

@ -65,6 +65,7 @@ This directory contains Discord slash commands for creating and managing voice c
- **Restart Resilience**: JSON file persistence survives bot restarts
- **Startup Verification**: Validates tracked channels still exist on bot startup
- **Graceful Error Handling**: Continues operation even if individual operations fail
- **Scorecard Cleanup**: Automatically unpublishes scorecards when associated voice channels are deleted
## Architecture
@ -123,6 +124,29 @@ if hasattr(self.bot, 'voice_cleanup_service'):
cleanup_service.tracker.add_channel(channel, channel_type, interaction.user.id)
```
### Scorecard Cleanup Integration
When a voice channel is cleaned up (deleted after being empty for the configured threshold), the cleanup service automatically unpublishes any scorecard associated with that voice channel's text channel. This prevents the live scorebug tracker from continuing to update scores for games that no longer have active voice channels.
**Cleanup Flow**:
1. Voice channel becomes empty and exceeds empty threshold
2. Cleanup service deletes the voice channel
3. Service checks if voice channel has associated `text_channel_id`
4. If found, unpublishes scorecard from that text channel
5. Live scorebug tracker stops updating that scorecard
**Integration Points**:
- `cleanup_service.py` imports `ScorecardTracker` from `commands.gameplay.scorecard_tracker`
- Scorecard unpublishing happens in three scenarios:
- Normal cleanup (channel deleted after being empty)
- Stale channel cleanup (channel already deleted externally)
- Startup verification (channel no longer exists when bot starts)
**Logging**:
```
✅ Cleaned up empty voice channel: Gameplay Phoenix (ID: 123456789)
📋 Unpublished scorecard from text channel 987654321 (voice channel cleanup)
```
### JSON Data Structure
```json
{
@ -135,12 +159,15 @@ if hasattr(self.bot, 'voice_cleanup_service'):
"created_at": "2025-01-15T10:30:00",
"last_checked": "2025-01-15T10:35:00",
"empty_since": "2025-01-15T10:32:00",
"creator_id": "111222333"
"creator_id": "111222333",
"text_channel_id": "555666777"
}
}
}
```
**Note**: The `text_channel_id` field links the voice channel to its associated text channel, enabling automatic scorecard cleanup when the voice channel is deleted.
## Configuration
### Cleanup Service Settings

11
commands/voice/channels.py
View File

@ -112,10 +112,17 @@ class VoiceChannelCommands(commands.Cog):
category=voice_category
)
# Add to cleanup service tracking
# Add to cleanup service tracking with text channel association
if hasattr(self.bot, 'voice_cleanup_service'):
cleanup_service = self.bot.voice_cleanup_service # type: ignore[attr-defined]
cleanup_service.tracker.add_channel(channel, channel_type, interaction.user.id)
self.logger.info(f"Adding voice channel {channel.name} (ID: {channel.id}) to tracking with text channel {interaction.channel_id}")
cleanup_service.tracker.add_channel(
channel,
channel_type,
interaction.user.id,
text_channel_id=interaction.channel_id # Associate with text channel
)
self.logger.info(f"Successfully added voice channel to tracking")
else:
self.logger.warning("Voice cleanup service not available, channel won't be tracked")

41
commands/voice/cleanup_service.py
View File

@ -10,6 +10,7 @@ import discord
from discord.ext import commands
from .tracker import VoiceChannelTracker
from commands.gameplay.scorecard_tracker import ScorecardTracker
logger = logging.getLogger(f'{__name__}.VoiceChannelCleanupService')
@ -23,6 +24,7 @@ class VoiceChannelCleanupService:
- Automatic empty channel cleanup
- Configurable cleanup intervals and thresholds
- Stale entry removal and recovery
- Automatic scorecard unpublishing when voice channel is cleaned up
"""
def __init__(self, data_file: str = "data/voice_channels.json"):
@ -33,6 +35,7 @@ class VoiceChannelCleanupService:
data_file: Path to the JSON data file for persistence
"""
self.tracker = VoiceChannelTracker(data_file)
self.scorecard_tracker = ScorecardTracker()
self.cleanup_interval = 60 # 5 minutes check interval
self.empty_threshold = 5 # Delete after 15 minutes empty
self._running = False
@ -111,10 +114,22 @@ class VoiceChannelCleanupService:
except (ValueError, TypeError):
pass
# Remove stale entries
# Remove stale entries and unpublish associated scorecards
for channel_id in channels_to_remove:
# Get channel data before removing to access text_channel_id
channel_data = self.tracker.get_tracked_channel(channel_id)
self.tracker.remove_channel(channel_id)
# Unpublish associated scorecard if it exists
if channel_data and channel_data.get("text_channel_id"):
try:
text_channel_id_int = int(channel_data["text_channel_id"])
was_unpublished = self.scorecard_tracker.unpublish_scorecard(text_channel_id_int)
if was_unpublished:
logger.info(f"📋 Unpublished scorecard from text channel {text_channel_id_int} (stale voice channel)")
except (ValueError, TypeError) as e:
logger.warning(f"Invalid text_channel_id in stale voice channel data: {e}")
# Also clean up any additional stale entries
stale_removed = self.tracker.cleanup_stale_entries(valid_channel_ids)
total_removed = len(channels_to_remove) + stale_removed
@ -238,10 +253,34 @@ class VoiceChannelCleanupService:
logger.info(f"✅ Cleaned up empty voice channel: {channel_name} (ID: {channel_id})")
# Unpublish associated scorecard if it exists
text_channel_id = channel_data.get("text_channel_id")
if text_channel_id:
try:
text_channel_id_int = int(text_channel_id)
was_unpublished = self.scorecard_tracker.unpublish_scorecard(text_channel_id_int)
if was_unpublished:
logger.info(f"📋 Unpublished scorecard from text channel {text_channel_id_int} (voice channel cleanup)")
else:
logger.debug(f"No scorecard found for text channel {text_channel_id_int}")
except (ValueError, TypeError) as e:
logger.warning(f"Invalid text_channel_id in voice channel data: {e}")
except discord.NotFound:
# Channel was already deleted
logger.info(f"Channel {channel_data.get('name', 'unknown')} was already deleted")
self.tracker.remove_channel(int(channel_data["channel_id"]))
# Still try to unpublish associated scorecard
text_channel_id = channel_data.get("text_channel_id")
if text_channel_id:
try:
text_channel_id_int = int(text_channel_id)
was_unpublished = self.scorecard_tracker.unpublish_scorecard(text_channel_id_int)
if was_unpublished:
logger.info(f"📋 Unpublished scorecard from text channel {text_channel_id_int} (stale voice channel cleanup)")
except (ValueError, TypeError) as e:
logger.warning(f"Invalid text_channel_id in voice channel data: {e}")
except discord.Forbidden:
logger.error(f"Missing permissions to delete channel {channel_data.get('name', 'unknown')}")
except Exception as e:

30
commands/voice/tracker.py
View File

@ -64,7 +64,8 @@ class VoiceChannelTracker:
self,
channel: discord.VoiceChannel,
channel_type: str,
creator_id: int
creator_id: int,
text_channel_id: Optional[int] = None
) -> None:
"""
Add a new channel to tracking.
@ -73,6 +74,7 @@ class VoiceChannelTracker:
channel: Discord voice channel object
channel_type: Type of channel ('public' or 'private')
creator_id: Discord user ID who created the channel
text_channel_id: Optional Discord text channel ID associated with this voice channel
"""
self._data.setdefault("voice_channels", {})[str(channel.id)] = {
"channel_id": str(channel.id),
@ -82,7 +84,8 @@ class VoiceChannelTracker:
"created_at": datetime.now(UTC).isoformat(),
"last_checked": datetime.now(UTC).isoformat(),
"empty_since": None,
"creator_id": str(creator_id)
"creator_id": str(creator_id),
"text_channel_id": str(text_channel_id) if text_channel_id else None
}
self.save_data()
logger.info(f"Added channel to tracking: {channel.name} (ID: {channel.id})")
@ -190,6 +193,29 @@ class VoiceChannelTracker:
channels = self._data.get("voice_channels", {})
return channels.get(str(channel_id))
def get_voice_channel_for_text_channel(self, text_channel_id: int) -> Optional[int]:
"""
Get voice channel ID associated with a text channel.
Args:
text_channel_id: Discord text channel ID
Returns:
Voice channel ID if found, None otherwise
"""
channels = self._data.get("voice_channels", {})
for voice_channel_id_str, channel_data in channels.items():
stored_text_channel_id = channel_data.get("text_channel_id")
if stored_text_channel_id:
try:
if int(stored_text_channel_id) == text_channel_id:
return int(voice_channel_id_str)
except (ValueError, TypeError):
continue
return None
def cleanup_stale_entries(self, valid_channel_ids: List[int]) -> int:
"""
Remove tracking entries for channels that no longer exist.

1
config.py
View File

@ -60,6 +60,7 @@ class BotConfig(BaseSettings):
# Base URLs
sba_base_url: str = "https://sba.manticorum.com"
sba_logo_url: str = f'{sba_base_url}/images/sba-logo.png'
# Application settings
log_level: str = "INFO"

191
services/scorebug_service.py Normal file
View File

@ -0,0 +1,191 @@
"""
Scorebug Service
Handles reading live game data from Google Sheets scorecards for real-time score displays.
"""
import asyncio
from typing import Dict, List, Any, Optional
import pygsheets
from utils.logging import get_contextual_logger
from exceptions import SheetsException
from services.sheets_service import SheetsService
class ScorebugData:
"""Data class for scorebug information."""
def __init__(self, data: Dict[str, Any]):
self.away_team_id = data.get('away_team_id', 1)
self.home_team_id = data.get('home_team_id', 1)
self.header = data.get('header', '')
self.away_score = data.get('away_score', 0)
self.home_score = data.get('home_score', 0)
self.which_half = data.get('which_half', '')
self.is_final = data.get('is_final', False)
self.runners = data.get('runners', [])
self.matchups = data.get('matchups', [])
self.summary = data.get('summary', [])
@property
def score_line(self) -> str:
"""Get formatted score line for display."""
return f"{self.away_score} @ {self.home_score}"
@property
def is_active(self) -> bool:
"""Check if game is currently active (not final)."""
return not self.is_final
class ScorebugService(SheetsService):
"""Google Sheets integration for reading live scorebug data."""
def __init__(self, credentials_path: Optional[str] = None):
"""
Initialize scorebug service.
Args:
credentials_path: Path to service account credentials JSON
"""
super().__init__(credentials_path)
self.logger = get_contextual_logger(f'{__name__}.ScorebugService')
async def read_scorebug_data(
self,
sheet_url_or_key: str,
full_length: bool = True
) -> ScorebugData:
"""
Read live scorebug data from Google Sheets scorecard.
Args:
sheet_url_or_key: Full URL or Google Sheets key
full_length: If True, includes summary data; if False, compact view
Returns:
ScorebugData object with game state
Raises:
SheetsException: If scorecard cannot be read
"""
try:
# Open scorecard
scorecard = await self.open_scorecard(sheet_url_or_key)
loop = asyncio.get_event_loop()
# Get Scorebug tab
scorebug_tab = await loop.run_in_executor(
None,
scorecard.worksheet_by_title,
'Scorebug'
)
# Read all data from B2:S20 for efficiency
all_data = await loop.run_in_executor(
None,
lambda: scorebug_tab.get_values('B2', 'S20', include_tailing_empty_rows=True)
)
self.logger.debug(f"Raw scorebug data (first 10 rows): {all_data[:10]}")
# Extract game state (B2:G8)
game_state = [
all_data[0][:6], all_data[1][:6], all_data[2][:6], all_data[3][:6],
all_data[4][:6], all_data[5][:6], all_data[6][:6]
]
self.logger.debug(f"Extracted game_state: {game_state}")
# Extract team IDs from game_state (already read from Scorebug tab)
# game_state[3] is away team row, game_state[4] is home team row
# First column (index 0) contains the team ID
try:
away_team_id = int(game_state[3][0]) if len(game_state) > 3 and len(game_state[3]) > 0 else None
home_team_id = int(game_state[4][0]) if len(game_state) > 4 and len(game_state[4]) > 0 else None
self.logger.debug(f"Parsed team IDs - Away: {away_team_id}, Home: {home_team_id}")
if away_team_id is None or home_team_id is None:
raise ValueError(f'Team IDs not found in scorebug (away: {away_team_id}, home: {home_team_id})')
except (ValueError, IndexError) as e:
self.logger.error(f"Failed to parse team IDs from scorebug: {e}")
raise ValueError(f'Could not extract team IDs from scorecard')
# Parse game state
header = game_state[0][0] if game_state[0] else ''
is_final = header[-5:] == 'FINAL' if header else False
self.logger.debug(f"Header: '{header}', Is Final: {is_final}")
self.logger.debug(f"Away team row (game_state[3]): {game_state[3] if len(game_state) > 3 else 'N/A'}")
self.logger.debug(f"Home team row (game_state[4]): {game_state[4] if len(game_state) > 4 else 'N/A'}")
# Parse scores with validation
try:
away_score_raw = game_state[3][2] if len(game_state) > 3 and len(game_state[3]) > 2 else '0'
self.logger.debug(f"Raw away score value: '{away_score_raw}'")
away_score = int(away_score_raw)
except (ValueError, IndexError) as e:
self.logger.warning(f"Failed to parse away score: {e}")
away_score = 0
try:
home_score_raw = game_state[4][2] if len(game_state) > 4 and len(game_state[4]) > 2 else '0'
self.logger.debug(f"Raw home score value: '{home_score_raw}'")
home_score = int(home_score_raw)
except (ValueError, IndexError) as e:
self.logger.warning(f"Failed to parse home score: {e}")
home_score = 0
which_half = game_state[3][4] if len(game_state) > 3 and len(game_state[3]) > 4 else ''
self.logger.debug(f"Parsed values - Away: {away_score}, Home: {home_score}, Which Half: '{which_half}'")
# Extract runners (K11:L14 → offset in all_data)
runners = [
all_data[9][9:11] if len(all_data) > 9 else [],
all_data[10][9:11] if len(all_data) > 10 else [],
all_data[11][9:11] if len(all_data) > 11 else [],
all_data[12][9:11] if len(all_data) > 12 else []
]
# Extract matchups if full_length (M11:N14 → offset in all_data)
matchups = []
if full_length:
matchups = [
all_data[9][11:13] if len(all_data) > 9 else [],
all_data[10][11:13] if len(all_data) > 10 else [],
all_data[11][11:13] if len(all_data) > 11 else [],
all_data[12][11:13] if len(all_data) > 12 else []
]
# Extract summary if full_length (Q11:R14 → offset in all_data)
summary = []
if full_length:
summary = [
all_data[9][15:17] if len(all_data) > 9 else [],
all_data[10][15:17] if len(all_data) > 10 else [],
all_data[11][15:17] if len(all_data) > 11 else [],
all_data[12][15:17] if len(all_data) > 12 else []
]
return ScorebugData({
'away_team_id': away_team_id,
'home_team_id': home_team_id,
'header': header,
'away_score': away_score,
'home_score': home_score,
'which_half': which_half,
'is_final': is_final,
'runners': runners,
'matchups': matchups,
'summary': summary
})
except pygsheets.WorksheetNotFound:
self.logger.error(f"Scorebug tab not found in scorecard")
raise SheetsException("Scorebug tab not found. Is this a valid scorecard?")
except Exception as e:
self.logger.error(f"Failed to read scorebug data: {e}")
raise SheetsException(f"Unable to read scorebug data: {str(e)}")

124
tasks/CLAUDE.md
View File

@ -78,7 +78,7 @@ async def _begin_freeze(self, current: Current):
See `services/CLAUDE.md` for complete service layer best practices.
### Base Task Pattern
All tasks follow a consistent structure:
All tasks follow a consistent structure with **MANDATORY** safe startup:
```python
from discord.ext import tasks
@ -105,12 +105,132 @@ class ExampleTask:
@task_loop.before_loop
async def before_task(self):
"""Wait for bot to be ready before starting."""
"""Wait for bot to be ready before starting - REQUIRED FOR SAFE STARTUP."""
await self.bot.wait_until_ready()
self.logger.info("Bot is ready, task starting")
```
### 🚨 CRITICAL: Safe Startup Pattern
**EVERY background task MUST use the `@task.before_loop` decorator with `await self.bot.wait_until_ready()`.**
This pattern prevents tasks from executing before:
- Discord connection is established
- Bot guilds are fully loaded
- Bot cache is populated
- Service dependencies are available
#### ✅ CORRECT Pattern (Always Use This)
```python
@tasks.loop(minutes=3)
async def my_task_loop(self):
"""Main task logic."""
# Your task code here
pass
@my_task_loop.before_loop
async def before_my_task(self):
"""Wait for bot to be ready before starting - REQUIRED."""
await self.bot.wait_until_ready()
self.logger.info("Bot is ready, my_task starting")
```
#### ❌ WRONG Pattern (Will Cause Errors)
```python
@tasks.loop(minutes=3)
async def my_task_loop(self):
"""Main task logic."""
# Task starts immediately - bot may not be ready!
# This will cause AttributeError, NoneType errors, etc.
pass
# Missing @before_loop - BAD!
```
#### Why This Is Critical
Without the `before_loop` pattern:
- **Guild lookup fails** - `bot.get_guild()` returns `None`
- **Channel lookup fails** - `guild.text_channels` is empty or incomplete
- **Cache errors** - Discord objects not fully populated
- **Service failures** - Dependencies may not be initialized
- **Race conditions** - Task runs before bot state is stable
#### Implementation Checklist
When creating a new task, ensure:
- [ ] `@tasks.loop()` decorator on main loop method
- [ ] `@task.before_loop` decorator on before method
- [ ] `await self.bot.wait_until_ready()` in before method
- [ ] Log message confirming task is ready to start
- [ ] Task started in `__init__()` with `self.task_loop.start()`
- [ ] Task cancelled in `cog_unload()` with `self.task_loop.cancel()`
## Current Tasks
### Live Scorebug Tracker (`live_scorebug_tracker.py`)
**Purpose:** Automated live game score updates for active games
**Schedule:** Every 3 minutes
**Operations:**
- **Live Scores Channel Update:**
- Reads all published scorecards from ScorecardTracker
- Generates compact scorebug embeds for active games
- Clears and updates `#live-sba-scores` channel
- Filters out final games (only shows active/in-progress)
- **Voice Channel Description Update:**
- For each active scorecard, checks for associated voice channel
- Updates voice channel topic with live score (e.g., "BOS 4 @ 3 NYY")
- Adds "- FINAL" suffix when game completes
- Gracefully handles missing or deleted voice channels
#### Key Features
- **Restart Resilience:** Uses JSON-based scorecard tracking
- **Voice Integration:** Bi-directional integration with voice channel system
- **Rate Limiting:** 1-second delay between scorecard reads
- **Error Resilience:** Continues operation despite individual failures
- **Safe Startup:** Uses `@before_loop` pattern with `await bot.wait_until_ready()`
#### Configuration
The tracker respects configuration settings:
```python
# config.py settings
guild_id: int # Target guild for operations
```
**Environment Variables:**
- `GUILD_ID` - Discord server ID
#### Scorecard Publishing
Users publish scorecards via `/publish-scorecard <url>`:
- Validates Google Sheets access and structure
- Stores text_channel_id → sheet_url mapping in JSON
- Persists across bot restarts
#### Voice Channel Association
When voice channels are created:
- Text channel ID stored in voice channel tracking data
- Enables scorebug → voice channel lookup
- Voice channel topic updated every 3 minutes with live scores
**Automatic Cleanup Integration:**
When voice channels are cleaned up (deleted after being empty):
- Voice cleanup service automatically unpublishes the associated scorecard
- Prevents live scorebug tracker from updating scores for games without active voice channels
- Ensures scorecard tracking stays synchronized with voice channel state
- Reduces unnecessary API calls to Google Sheets for inactive games
#### Channel Requirements
- **#live-sba-scores** - Live scorebug display channel
#### Error Handling
- Comprehensive try/catch blocks with structured logging
- Graceful degradation if channels not found
- Silent skip for deleted voice channels
- Prevents duplicate error messages
- Continues operation despite individual scorecard failures
### Transaction Freeze/Thaw (`transaction_freeze.py`)
**Purpose:** Automated weekly system for freezing transactions and processing contested player acquisitions

323
tasks/live_scorebug_tracker.py Normal file
View File

@ -0,0 +1,323 @@
"""
Live Scorebug Tracker
Background task that monitors published scorecards and updates live score displays.
"""
import asyncio
from typing import List, Optional
import discord
from discord.ext import tasks, commands
from models.team import Team
from utils.logging import get_contextual_logger
from services.scorebug_service import ScorebugData, ScorebugService
from services.team_service import team_service
from commands.gameplay.scorecard_tracker import ScorecardTracker
from commands.voice.tracker import VoiceChannelTracker
from views.embeds import EmbedTemplate, EmbedColors
from config import get_config
from exceptions import SheetsException
class LiveScorebugTracker:
"""
Manages live scorebug updates for active games.
Features:
- Updates live scores channel every 3 minutes
- Updates voice channel descriptions with live scores
- Clears displays when no active games
- Error resilient with graceful degradation
"""
def __init__(self, bot: commands.Bot):
"""
Initialize the live scorebug tracker.
Args:
bot: Discord bot instance
"""
self.bot = bot
self.logger = get_contextual_logger(f'{__name__}.LiveScorebugTracker')
self.scorebug_service = ScorebugService()
self.scorecard_tracker = ScorecardTracker()
self.voice_tracker = VoiceChannelTracker()
# Start the monitoring loop
self.update_loop.start()
self.logger.info("Live scorebug tracker initialized")
def cog_unload(self):
"""Stop the task when service is unloaded."""
self.update_loop.cancel()
self.logger.info("Live scorebug tracker stopped")
@tasks.loop(minutes=3)
async def update_loop(self):
"""
Main update loop - runs every 3 minutes.
Updates:
- Live scores channel with all active scorebugs
- Voice channel descriptions with live scores
"""
try:
await self._update_scorebugs()
except Exception as e:
self.logger.error(f"Error in scorebug update loop: {e}", exc_info=True)
@update_loop.before_loop
async def before_update_loop(self):
"""Wait for bot to be ready before starting."""
await self.bot.wait_until_ready()
self.logger.info("Live scorebug tracker ready to start monitoring")
async def _update_scorebugs(self):
"""Update all scorebug displays."""
config = get_config()
guild = self.bot.get_guild(config.guild_id)
if not guild:
self.logger.warning(f"Guild {config.guild_id} not found, skipping update")
return
# Get live scores channel
live_scores_channel = discord.utils.get(guild.text_channels, name='live-sba-scores')
if not live_scores_channel:
self.logger.warning("live-sba-scores channel not found, skipping channel update")
# Don't return - still update voice channels
else:
# Get all published scorecards
all_scorecards = self.scorecard_tracker.get_all_scorecards()
if not all_scorecards:
# No active scorebugs - clear the channel
await self._clear_live_scores_channel(live_scores_channel)
return
# Read all scorebugs and create embeds
active_scorebugs = []
for text_channel_id, sheet_url in all_scorecards:
try:
scorebug_data = await self.scorebug_service.read_scorebug_data(
sheet_url,
full_length=False # Compact view for live channel
)
# Only include active (non-final) games
if scorebug_data.is_active:
# Get team data
away_team = await team_service.get_team(scorebug_data.away_team_id)
home_team = await team_service.get_team(scorebug_data.home_team_id)
if away_team is None or home_team is None:
raise ValueError(f'Error looking up teams in scorecard; IDs provided: {scorebug_data.away_team_id} & {scorebug_data.home_team_id}')
# Create compact embed
embed = await self._create_compact_scorebug_embed(
scorebug_data,
away_team,
home_team
)
active_scorebugs.append(embed)
# Update associated voice channel if it exists
await self._update_voice_channel_description(
text_channel_id,
scorebug_data,
away_team,
home_team
)
await asyncio.sleep(1) # Rate limit between reads
except SheetsException as e:
self.logger.warning(f"Could not read scorecard {sheet_url}: {e}")
except Exception as e:
self.logger.error(f"Error processing scorecard {sheet_url}: {e}")
# Update live scores channel
if active_scorebugs:
await self._post_scorebugs_to_channel(live_scores_channel, active_scorebugs)
else:
# All games finished - clear the channel
await self._clear_live_scores_channel(live_scores_channel)
async def _create_compact_scorebug_embed(
self,
scorebug_data,
away_team: Team,
home_team: Team
) -> discord.Embed:
"""
Create a compact scorebug embed for the live channel.
Args:
scorebug_data: ScorebugData object
away_team: Away team object (optional)
home_team: Home team object (optional)
Returns:
Discord embed with compact scorebug
"""
# Determine winning team for embed color
if scorebug_data.away_score > scorebug_data.home_score and away_team:
embed_color = away_team.get_color_int()
elif scorebug_data.home_score > scorebug_data.away_score and home_team:
embed_color = home_team.get_color_int()
else:
embed_color = EmbedColors.INFO
# Create compact embed
embed = discord.Embed(
title=scorebug_data.header,
color=embed_color
)
# Add score
away_abbrev = away_team.abbrev if away_team else "AWAY"
home_abbrev = home_team.abbrev if home_team else "HOME"
score_text = (
f"```\n"
f"{away_abbrev:<4} {scorebug_data.away_score:>2}\n"
f"{home_abbrev:<4} {scorebug_data.home_score:>2}\n"
f"```"
)
embed.add_field(
name="Score",
value=score_text,
inline=True
)
embed.add_field(
name="Status",
value=f"**{scorebug_data.which_half}**",
inline=True
)
return embed
async def _post_scorebugs_to_channel(
self,
channel: discord.TextChannel,
embeds: List[discord.Embed]
):
"""
Post scorebugs to the live scores channel.
Args:
channel: Discord text channel
embeds: List of scorebug embeds
"""
try:
# Clear old messages
async for message in channel.history(limit=25):
await message.delete()
# Post new scorebugs (Discord allows up to 10 embeds per message)
if len(embeds) <= 10:
await channel.send(embeds=embeds)
else:
# Split into multiple messages if more than 10 embeds
for i in range(0, len(embeds), 10):
batch = embeds[i:i+10]
await channel.send(embeds=batch)
self.logger.info(f"Posted {len(embeds)} scorebugs to live-sba-scores")
except discord.Forbidden:
self.logger.error("Missing permissions to update live-sba-scores channel")
except Exception as e:
self.logger.error(f"Error posting scorebugs: {e}")
async def _clear_live_scores_channel(self, channel: discord.TextChannel):
"""
Clear the live scores channel when no active games.
Args:
channel: Discord text channel
"""
try:
# Clear all messages
async for message in channel.history(limit=25):
await message.delete()
self.logger.info("Cleared live-sba-scores channel (no active games)")
except discord.Forbidden:
self.logger.error("Missing permissions to clear live-sba-scores channel")
except Exception as e:
self.logger.error(f"Error clearing channel: {e}")
async def _update_voice_channel_description(
self,
text_channel_id: int,
scorebug_data: ScorebugData,
away_team: Team,
home_team: Team
):
"""
Update voice channel description with live score.
Args:
text_channel_id: Text channel ID where scorecard was published
scorebug_data: ScorebugData object
away_team: Away team object (optional)
home_team: Home team object (optional)
"""
try:
# Check if there's an associated voice channel
voice_channel_id = self.voice_tracker.get_voice_channel_for_text_channel(text_channel_id)
if not voice_channel_id:
self.logger.debug(f'No voice channel associated with text channel ID {text_channel_id} (may have been cleaned up)')
return # No associated voice channel
# Get the voice channel
config = get_config()
guild = self.bot.get_guild(config.guild_id)
if not guild:
return
voice_channel = guild.get_channel(voice_channel_id)
if not voice_channel or not isinstance(voice_channel, discord.VoiceChannel):
self.logger.debug(f"Voice channel {voice_channel_id} not found or wrong type")
return
# Format description: "BOS 4 @ 3 NYY" or "BOS 5 @ 3 NYY - FINAL"
away_abbrev = away_team.abbrev if away_team else "AWAY"
home_abbrev = home_team.abbrev if home_team else "HOME"
if scorebug_data.is_final:
description = f"{away_abbrev} {scorebug_data.away_score} @ {scorebug_data.home_score} {home_abbrev} - FINAL"
else:
description = f"{away_abbrev} {scorebug_data.away_score} @ {scorebug_data.home_score} {home_abbrev}"
# Update voice channel description (topic)
await voice_channel.edit(status=description)
self.logger.debug(f"Updated voice channel {voice_channel.name} description to: {description}")
except discord.Forbidden:
self.logger.warning(f"Missing permissions to update voice channel {voice_channel_id}")
except Exception as e:
self.logger.error(f"Error updating voice channel description: {e}")
def setup_scorebug_tracker(bot: commands.Bot) -> LiveScorebugTracker:
"""
Setup function to initialize the live scorebug tracker.
Args:
bot: Discord bot instance
Returns:
LiveScorebugTracker instance
"""
return LiveScorebugTracker(bot)

92
tests/test_commands_voice.py
View File

@ -292,6 +292,98 @@ class TestVoiceChannelCleanupService:
# Should have removed from tracking
assert "123" not in cleanup_service.tracker._data["voice_channels"]
@pytest.mark.asyncio
async def test_cleanup_channel_with_scorecard(self, cleanup_service, mock_bot):
"""Test that cleaning up a channel also unpublishes associated scorecard."""
# Mock guild and channel
mock_guild = MagicMock()
mock_guild.id = 999
mock_channel = AsyncMock(spec=discord.VoiceChannel)
mock_channel.id = 123
mock_channel.members = [] # Empty channel
mock_bot.get_guild.return_value = mock_guild
mock_guild.get_channel.return_value = mock_channel
# Add channel to tracking with associated text channel
cleanup_service.tracker._data["voice_channels"]["123"] = {
"channel_id": "123",
"guild_id": "999",
"name": "Test Channel",
"text_channel_id": "555"
}
# Add a scorecard for the text channel
cleanup_service.scorecard_tracker._data = {
"scorecards": {
"555": {
"text_channel_id": "555",
"sheet_url": "https://example.com/sheet",
"published_at": "2025-01-15T10:30:00",
"publisher_id": "12345"
}
}
}
channel_data = {
"channel_id": "123",
"guild_id": "999",
"name": "Test Channel",
"text_channel_id": "555"
}
await cleanup_service.cleanup_channel(mock_bot, channel_data)
# Should have deleted the channel
mock_channel.delete.assert_called_once_with(reason="Automatic cleanup - empty for 15+ minutes")
# Should have removed from voice channel tracking
assert "123" not in cleanup_service.tracker._data["voice_channels"]
# Should have unpublished the scorecard
assert "555" not in cleanup_service.scorecard_tracker._data["scorecards"]
@pytest.mark.asyncio
async def test_verify_tracked_channels_unpublishes_scorecards(self, cleanup_service, mock_bot):
"""Test that verifying tracked channels also unpublishes associated scorecards."""
# Add test data with a stale voice channel that has an associated scorecard
cleanup_service.tracker._data = {
"voice_channels": {
"123": {
"channel_id": "123",
"guild_id": "999",
"name": "Stale Channel",
"text_channel_id": "555"
}
}
}
# Add a scorecard for the text channel
cleanup_service.scorecard_tracker._data = {
"scorecards": {
"555": {
"text_channel_id": "555",
"sheet_url": "https://example.com/sheet",
"published_at": "2025-01-15T10:30:00",
"publisher_id": "12345"
}
}
}
# Mock guild but not the channel (simulating deleted channel)
mock_guild = MagicMock()
mock_guild.id = 999
mock_bot.get_guild.return_value = mock_guild
mock_guild.get_channel.return_value = None # Channel no longer exists
await cleanup_service.verify_tracked_channels(mock_bot)
# Voice channel should be removed from tracking
assert "123" not in cleanup_service.tracker._data["voice_channels"]
# Scorecard should be unpublished
assert "555" not in cleanup_service.scorecard_tracker._data["scorecards"]
class TestVoiceChannelCommands:
"""Test voice channel command functionality."""

92
utils/text_utils.py Normal file
View File

@ -0,0 +1,92 @@
"""
Text Utility Functions
Provides text manipulation and formatting utilities for Discord bot operations.
"""
def split_text_for_fields(text: str, max_length: int = 1024, split_on: str = '\n') -> list[str]:
"""
Split text into chunks that fit Discord field value limits.
Discord embeds have a field value limit of 1024 characters. This function intelligently
splits longer text into multiple chunks while preserving readability by splitting on
semantic boundaries (default: newlines).
Args:
text: Text to split into chunks
max_length: Maximum characters per chunk (default 1024 for Discord field values)
split_on: Character/string to split on for clean breaks (default newline)
Returns:
List of text chunks, each under max_length characters
Examples:
>>> short_text = "This is short"
>>> split_text_for_fields(short_text)
['This is short']
>>> long_text = "Line 1\\nLine 2\\nLine 3\\n..." * 100
>>> chunks = split_text_for_fields(long_text, max_length=100)
>>> all(len(chunk) <= 100 for chunk in chunks)
True
>>> # Custom delimiter
>>> text = "Part 1. Part 2. Part 3."
>>> split_text_for_fields(text, max_length=10, split_on='. ')
['Part 1.', 'Part 2.', 'Part 3.']
Notes:
- If text is already under max_length, returns single-item list
- Splits on boundaries to preserve formatting (no mid-line breaks)
- Trailing delimiters are removed from final chunks
- Empty segments are preserved if they exist in original text
"""
# Handle edge cases
if not text:
return ['']
if len(text) <= max_length:
return [text]
chunks = []
current_chunk = []
current_length = 0
# Split on the delimiter (e.g., '\n')
segments = text.split(split_on)
for i, segment in enumerate(segments):
# Add delimiter back except for last segment
is_last_segment = (i == len(segments) - 1)
segment_with_delimiter = segment if is_last_segment else segment + split_on
segment_length = len(segment_with_delimiter)
# If adding this segment would exceed limit, start new chunk
if current_length + segment_length > max_length and current_chunk:
# Save current chunk
chunks.append(''.join(current_chunk).rstrip(split_on))
current_chunk = []
current_length = 0
# Handle edge case: single segment longer than max_length
if segment_length > max_length:
# Split mid-segment as last resort (shouldn't happen with reasonable max_length)
if current_chunk:
chunks.append(''.join(current_chunk).rstrip(split_on))
current_chunk = []
current_length = 0
# Split the long segment into character chunks
for j in range(0, len(segment_with_delimiter), max_length):
chunks.append(segment_with_delimiter[j:j + max_length])
else:
# Add segment to current chunk
current_chunk.append(segment_with_delimiter)
current_length += segment_length
# Add remaining chunk if any
if current_chunk:
chunks.append(''.join(current_chunk).rstrip(split_on))
return chunks

14
views/players.py
View File

@ -148,7 +148,7 @@ class PlayerStatsView(BaseView):
show_pitching=self.show_pitching)
except Exception as e:
self.logger.error("Failed to update embed", error=str(e), exc_info=True)
self.logger.error("Failed to update embed", error=e, exc_info=True)
# Try to send error message
try:
@ -172,12 +172,10 @@ class PlayerStatsView(BaseView):
# Determine embed color based on team
embed_color = EmbedColors.PRIMARY
if hasattr(player, 'team') and player.team and hasattr(player.team, 'color'):
try:
# Convert hex color string to int
embed_color = int(player.team.color, 16)
except (ValueError, TypeError):
embed_color = EmbedColors.PRIMARY
if player.team and player.team.color:
embed_color = int(player.team.color, 16)
else:
embed_color = EmbedColors.PRIMARY
# Create base embed with player name as title
# Add injury indicator emoji if player is injured
@ -204,7 +202,7 @@ class PlayerStatsView(BaseView):
# Add Major League affiliate if this is a Minor League team
if player.team.roster_type() == RosterType.MINOR_LEAGUE:
major_affiliate = player.team.get_major_league_affiliate()
major_affiliate = player.team.major_league_affiliate()
if major_affiliate:
embed.add_field(
name="Major Affiliate",