9770e360c3
- Modified cogs/dice.py, cogs/fun.py, db_calls.py - Added COMMAND_LIST.md, api_calls/custom_command.py, sba_is_fun.db Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
123 lines
3.6 KiB
Python
123 lines
3.6 KiB
Python
import logging
|
|
import pydantic
|
|
|
|
from typing import Optional, List
|
|
from datetime import datetime
|
|
from db_calls import db_get
|
|
from exceptions import log_exception, ApiException
|
|
|
|
logger = logging.getLogger('discord_app')
|
|
|
|
|
|
class CustomCommandCreator(pydantic.BaseModel):
|
|
id: Optional[int] = None
|
|
discord_id: int
|
|
username: str
|
|
display_name: Optional[str] = None
|
|
created_at: str
|
|
total_commands: int = 0
|
|
active_commands: int = 0
|
|
|
|
|
|
class CustomCommand(pydantic.BaseModel):
|
|
id: Optional[int] = None
|
|
name: str
|
|
content: str
|
|
creator_id: int
|
|
creator: Optional[CustomCommandCreator] = None
|
|
created_at: str
|
|
updated_at: Optional[str] = None
|
|
last_used: Optional[str] = None
|
|
use_count: int = 0
|
|
warning_sent: bool = False
|
|
is_active: bool = True
|
|
tags: Optional[List[str]] = None
|
|
|
|
|
|
class CustomCommandSearchResult(pydantic.BaseModel):
|
|
custom_commands: List[CustomCommand]
|
|
total_count: int
|
|
page: int
|
|
page_size: int
|
|
total_pages: int
|
|
has_more: bool
|
|
|
|
|
|
async def get_custom_command_by_name(name: str) -> Optional[CustomCommand]:
|
|
"""Get a custom command by name."""
|
|
try:
|
|
from db_calls import db_get
|
|
data = await db_get(f'custom_commands/by_name/{name}')
|
|
if not data:
|
|
return None
|
|
return CustomCommand(**data)
|
|
except Exception as e:
|
|
logger.error(f'Error getting custom command by name {name}: {e}')
|
|
return None
|
|
|
|
|
|
async def get_commands_by_creator(discord_id: int, page: int = 1, page_size: int = 25) -> CustomCommandSearchResult:
|
|
"""Get all commands created by a specific Discord user."""
|
|
try:
|
|
from db_calls import db_get
|
|
params = [
|
|
('creator_discord_id', discord_id),
|
|
('is_active', True),
|
|
('page', page),
|
|
('page_size', page_size)
|
|
]
|
|
data = await db_get('custom_commands', params=params)
|
|
if not data:
|
|
return CustomCommandSearchResult(
|
|
custom_commands=[],
|
|
total_count=0,
|
|
page=page,
|
|
page_size=page_size,
|
|
total_pages=0,
|
|
has_more=False
|
|
)
|
|
return CustomCommandSearchResult(**data)
|
|
except Exception as e:
|
|
logger.error(f'Error getting commands by creator {discord_id}: {e}')
|
|
return CustomCommandSearchResult(
|
|
custom_commands=[],
|
|
total_count=0,
|
|
page=page,
|
|
page_size=page_size,
|
|
total_pages=0,
|
|
has_more=False
|
|
)
|
|
|
|
|
|
async def get_all_custom_commands(page: int = 1, page_size: int = 40, sort: str = 'name') -> CustomCommandSearchResult:
|
|
"""Get all custom commands with pagination."""
|
|
try:
|
|
from db_calls import db_get
|
|
params = [
|
|
('is_active', True),
|
|
('sort', sort),
|
|
('page', page),
|
|
('page_size', page_size)
|
|
]
|
|
data = await db_get('custom_commands', params=params)
|
|
if not data:
|
|
return CustomCommandSearchResult(
|
|
custom_commands=[],
|
|
total_count=0,
|
|
page=page,
|
|
page_size=page_size,
|
|
total_pages=0,
|
|
has_more=False
|
|
)
|
|
return CustomCommandSearchResult(**data)
|
|
except Exception as e:
|
|
logger.error(f'Error getting all custom commands: {e}')
|
|
return CustomCommandSearchResult(
|
|
custom_commands=[],
|
|
total_count=0,
|
|
page=page,
|
|
page_size=page_size,
|
|
total_pages=0,
|
|
has_more=False
|
|
)
|