45 lines
1.1 KiB
Python
45 lines
1.1 KiB
Python
"""JARVIS task utilities."""
|
|
import asyncio
|
|
from datetime import datetime, timezone
|
|
from logging import Logger
|
|
from typing import Coroutine
|
|
|
|
from naff.models.discord.embed import Embed
|
|
|
|
|
|
async def runat(when: datetime, coro: Coroutine, logger: Logger) -> None:
|
|
"""
|
|
Run a task at a scheduled time.
|
|
|
|
Args:
|
|
when: When to run the task
|
|
coro: Coroutine to execute
|
|
logger: Global logger
|
|
"""
|
|
logger.debug(f"Scheduling task {coro.__name__} for {when.isoformat()}")
|
|
delay = when - datetime.now(tz=timezone.utc)
|
|
await asyncio.sleep(delay.total_seconds())
|
|
await coro
|
|
|
|
|
|
def build_embed(
|
|
title: str,
|
|
description: str,
|
|
fields: list,
|
|
color: str = "#FF0000",
|
|
timestamp: datetime = None,
|
|
**kwargs: dict,
|
|
) -> Embed:
|
|
"""Embed builder utility function."""
|
|
if not timestamp:
|
|
timestamp = datetime.now()
|
|
embed = Embed(
|
|
title=title,
|
|
description=description,
|
|
color=color,
|
|
timestamp=timestamp,
|
|
**kwargs,
|
|
)
|
|
for field in fields:
|
|
embed.add_field(**field.to_dict())
|
|
return embed
|