mirror of https://github.com/stijndcl/didier
Compare commits
5 Commits
6f0ac487cc
...
2de75fd168
| Author | SHA1 | Date |
|---|---|---|
|
|
2de75fd168 | |
|
|
e1af53cf44 | |
|
|
b581c3e5dc | |
|
|
0186a0793a | |
|
|
654fbcd46b |
|
|
@ -27,10 +27,13 @@ class Currency(commands.Cog):
|
|||
|
||||
@commands.command(name="Award")
|
||||
@commands.check(is_owner)
|
||||
async def award(self, ctx: commands.Context, user: discord.User, amount: abbreviated_number): # type: ignore
|
||||
async def award(
|
||||
self,
|
||||
ctx: commands.Context,
|
||||
user: discord.User,
|
||||
amount: typing.Annotated[int, abbreviated_number],
|
||||
):
|
||||
"""Award a user a given amount of Didier Dinks"""
|
||||
amount = typing.cast(int, amount)
|
||||
|
||||
async with self.client.postgres_session as session:
|
||||
await crud.add_dinks(session, user.id, amount)
|
||||
plural = pluralize("Didier Dink", amount)
|
||||
|
|
@ -45,8 +48,7 @@ class Currency(commands.Cog):
|
|||
async with self.client.postgres_session as session:
|
||||
bank = await crud.get_bank(session, ctx.author.id)
|
||||
|
||||
embed = discord.Embed(colour=discord.Colour.blue())
|
||||
embed.set_author(name=f"{ctx.author.display_name}'s Bank")
|
||||
embed = discord.Embed(title=f"{ctx.author.display_name}'s Bank", colour=discord.Colour.blue())
|
||||
embed.set_thumbnail(url=ctx.author.avatar.url)
|
||||
|
||||
embed.add_field(name="Interest level", value=bank.interest_level)
|
||||
|
|
@ -61,8 +63,7 @@ class Currency(commands.Cog):
|
|||
async with self.client.postgres_session as session:
|
||||
bank = await crud.get_bank(session, ctx.author.id)
|
||||
|
||||
embed = discord.Embed(colour=discord.Colour.blue())
|
||||
embed.set_author(name="Bank upgrades")
|
||||
embed = discord.Embed(title="Bank upgrades", colour=discord.Colour.blue())
|
||||
|
||||
embed.add_field(
|
||||
name=f"Interest ({bank.interest_level})", value=str(interest_upgrade_price(bank.interest_level))
|
||||
|
|
@ -118,10 +119,8 @@ class Currency(commands.Cog):
|
|||
await ctx.reply(f"**{ctx.author.display_name}** has **{bank.dinks}** {plural}.", mention_author=False)
|
||||
|
||||
@commands.command(name="Invest", aliases=["Deposit", "Dep"])
|
||||
async def invest(self, ctx: commands.Context, amount: abbreviated_number): # type: ignore
|
||||
async def invest(self, ctx: commands.Context, amount: typing.Annotated[typing.Union[str, int], abbreviated_number]):
|
||||
"""Invest a given amount of Didier Dinks"""
|
||||
amount = typing.cast(typing.Union[str, int], amount)
|
||||
|
||||
async with self.client.postgres_session as session:
|
||||
invested = await crud.invest(session, ctx.author.id, amount)
|
||||
plural = pluralize("Didier Dink", invested)
|
||||
|
|
|
|||
|
|
@ -10,13 +10,12 @@ from didier import Didier
|
|||
class CustomHelpCommand(commands.MinimalHelpCommand):
|
||||
"""Customised Help command to override the default implementation
|
||||
|
||||
The default is ugly as hell so we do some fiddling with it
|
||||
The default is ugly as hell, so we do some fiddling with it
|
||||
"""
|
||||
|
||||
def _help_embed_base(self, title: str) -> discord.Embed:
|
||||
"""Create the base structure for the embeds that get sent with the Help commands"""
|
||||
embed = discord.Embed(colour=discord.Colour.blue())
|
||||
embed.set_author(name=title)
|
||||
embed = discord.Embed(title=title, colour=discord.Colour.blue())
|
||||
embed.set_footer(text="Syntax: Didier Help [Categorie] of Didier Help [Commando]")
|
||||
return embed
|
||||
|
||||
|
|
|
|||
|
|
@ -1,3 +1,6 @@
|
|||
from datetime import date
|
||||
from typing import Optional
|
||||
|
||||
import discord
|
||||
from discord import app_commands
|
||||
from discord.ext import commands
|
||||
|
|
@ -5,7 +8,11 @@ from discord.ext import commands
|
|||
from database.crud import ufora_courses
|
||||
from database.crud.deadlines import get_deadlines
|
||||
from didier import Didier
|
||||
from didier.data.apis.hydra import fetch_menu
|
||||
from didier.data.embeds.deadlines import Deadlines
|
||||
from didier.data.embeds.hydra import no_menu_found
|
||||
from didier.exceptions import HTTPException
|
||||
from didier.utils.discord.converters.time import DateTransformer
|
||||
from didier.utils.discord.flags.school import StudyGuideFlags
|
||||
|
||||
|
||||
|
|
@ -26,6 +33,25 @@ class School(commands.Cog):
|
|||
embed = Deadlines(deadlines).to_embed()
|
||||
await ctx.reply(embed=embed, mention_author=False, ephemeral=False)
|
||||
|
||||
@commands.hybrid_command(
|
||||
name="menu",
|
||||
description="Show the menu in the Ghent University restaurants.",
|
||||
aliases=["Eten", "Food"],
|
||||
)
|
||||
@app_commands.rename(day_dt="date")
|
||||
async def menu(self, ctx: commands.Context, day_dt: Optional[app_commands.Transform[date, DateTransformer]] = None):
|
||||
"""Show the menu in the Ghent University restaurants.
|
||||
|
||||
Menus are Dutch, as a lot of dishes have very weird translations
|
||||
"""
|
||||
async with ctx.typing():
|
||||
try:
|
||||
menu = await fetch_menu(self.client.http_session, day_dt)
|
||||
embed = menu.to_embed(day_dt=day_dt)
|
||||
except HTTPException:
|
||||
embed = no_menu_found(day_dt)
|
||||
await ctx.reply(embed=embed, mention_author=False)
|
||||
|
||||
@commands.hybrid_command(
|
||||
name="fiche", description="Sends the link to the study guide for [Course]", aliases=["guide", "studiefiche"]
|
||||
)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,15 @@
|
|||
from datetime import date
|
||||
|
||||
from aiohttp import ClientSession
|
||||
|
||||
from didier.data.embeds.hydra import Menu
|
||||
from didier.utils.http.requests import ensure_get
|
||||
|
||||
__all__ = ["fetch_menu"]
|
||||
|
||||
|
||||
async def fetch_menu(http_session: ClientSession, day_dt: date) -> Menu:
|
||||
"""Fetch the menu for a given day"""
|
||||
endpoint = f"https://hydra.ugent.be/api/2.0/resto/menu/nl/{day_dt.year}/{day_dt.month}/{day_dt.day}.json"
|
||||
async with ensure_get(http_session, endpoint, log_exceptions=False) as response:
|
||||
return Menu.parse_obj(response)
|
||||
|
|
@ -13,7 +13,7 @@ class EmbedBaseModel(ABC):
|
|||
"""Abstract base class for a model that can be turned into a Discord embed"""
|
||||
|
||||
@abstractmethod
|
||||
def to_embed(self, **kwargs: dict) -> discord.Embed:
|
||||
def to_embed(self, **kwargs) -> discord.Embed:
|
||||
"""Turn this model into a Discord embed"""
|
||||
raise NotImplementedError
|
||||
|
||||
|
|
|
|||
|
|
@ -22,9 +22,8 @@ class Deadlines(EmbedBaseModel):
|
|||
self.deadlines.sort(key=lambda deadline: deadline.deadline)
|
||||
|
||||
@overrides
|
||||
def to_embed(self, **kwargs: dict) -> discord.Embed:
|
||||
embed = discord.Embed(colour=discord.Colour.dark_gold())
|
||||
embed.set_author(name="Upcoming Deadlines")
|
||||
def to_embed(self, **kwargs) -> discord.Embed:
|
||||
embed = discord.Embed(title="Upcoming Deadlines", colour=discord.Colour.dark_gold())
|
||||
now = tz_aware_now()
|
||||
|
||||
has_active_deadlines = False
|
||||
|
|
|
|||
|
|
@ -37,8 +37,7 @@ def create_error_embed(ctx: commands.Context, exception: Exception) -> discord.E
|
|||
|
||||
invocation = f"{ctx.author.display_name} in {origin}"
|
||||
|
||||
embed = discord.Embed(colour=discord.Colour.red())
|
||||
embed.set_author(name="Error")
|
||||
embed = discord.Embed(title="Error", colour=discord.Colour.red())
|
||||
embed.add_field(name="Command", value=f"{ctx.message.content}", inline=True)
|
||||
embed.add_field(name="Context", value=invocation, inline=True)
|
||||
embed.add_field(name="Exception", value=abbreviate(str(exception), Limits.EMBED_FIELD_VALUE_LENGTH), inline=False)
|
||||
|
|
|
|||
|
|
@ -19,8 +19,7 @@ class GoogleSearch(EmbedBaseModel):
|
|||
|
||||
def _error_embed(self) -> discord.Embed:
|
||||
"""Custom embed for unsuccessful requests"""
|
||||
embed = discord.Embed(colour=discord.Colour.red())
|
||||
embed.set_author(name="Google Search")
|
||||
embed = discord.Embed(title="Google Search", colour=discord.Colour.red())
|
||||
|
||||
# Empty embed
|
||||
if not self.data.results:
|
||||
|
|
@ -33,12 +32,11 @@ class GoogleSearch(EmbedBaseModel):
|
|||
return embed
|
||||
|
||||
@overrides
|
||||
def to_embed(self, **kwargs: dict) -> discord.Embed:
|
||||
def to_embed(self, **kwargs) -> discord.Embed:
|
||||
if not self.data.results or self.data.status_code != HTTPStatus.OK:
|
||||
return self._error_embed()
|
||||
|
||||
embed = discord.Embed(colour=discord.Colour.blue())
|
||||
embed.set_author(name="Google Search")
|
||||
embed = discord.Embed(title="Google Search", colour=discord.Colour.blue())
|
||||
embed.set_footer(text=self.data.result_stats or None)
|
||||
|
||||
# Add all results into the description
|
||||
|
|
|
|||
|
|
@ -0,0 +1,3 @@
|
|||
from .menu import Menu, no_menu_found
|
||||
|
||||
__all__ = ["Menu", "no_menu_found"]
|
||||
|
|
@ -0,0 +1,126 @@
|
|||
from datetime import date
|
||||
from enum import Enum
|
||||
from typing import Optional, cast
|
||||
|
||||
import discord
|
||||
from overrides import overrides
|
||||
from pydantic import BaseModel
|
||||
|
||||
from didier.data.embeds.base import EmbedPydantic
|
||||
from didier.utils.discord.colours import ghent_university_blue, ghent_university_yellow
|
||||
from didier.utils.types.datetime import int_to_weekday
|
||||
from didier.utils.types.string import leading
|
||||
|
||||
__all__ = ["Menu", "no_menu_found"]
|
||||
|
||||
|
||||
class _MealKind(str, Enum):
|
||||
FISH = "fish"
|
||||
MEAT = "meat"
|
||||
SOUP = "soup"
|
||||
VEGAN = "vegan"
|
||||
VEGETARIAN = "vegetarian"
|
||||
|
||||
|
||||
class _MealType(str, Enum):
|
||||
COLD = "cold"
|
||||
MAIN = "main"
|
||||
SIDE = "side"
|
||||
|
||||
|
||||
class _Meal(BaseModel):
|
||||
"""Model for an item on the menu"""
|
||||
|
||||
kind: _MealKind
|
||||
name: str
|
||||
price: str
|
||||
type: _MealType
|
||||
|
||||
|
||||
class Menu(EmbedPydantic):
|
||||
"""Embed that shows the menu in Ghent University restaurants"""
|
||||
|
||||
meals: list[_Meal] = []
|
||||
open: bool
|
||||
vegetables: list[str] = []
|
||||
message: Optional[str] = None
|
||||
|
||||
def _get_dutch_meal_prefix(self, meal: _Meal) -> str:
|
||||
if meal.kind == _MealKind.MEAT:
|
||||
prefix = "Vlees"
|
||||
elif meal.kind == _MealKind.FISH:
|
||||
prefix = "Vis"
|
||||
elif meal.kind == _MealKind.VEGETARIAN:
|
||||
prefix = "Vegetarisch"
|
||||
else:
|
||||
prefix = "Vegan"
|
||||
|
||||
return prefix
|
||||
|
||||
def _get_soups(self) -> str:
|
||||
acc = ""
|
||||
|
||||
for meal in self.meals:
|
||||
if meal.kind == _MealKind.SOUP:
|
||||
acc += f"{meal.name} ({meal.price})\n"
|
||||
|
||||
return acc.strip()
|
||||
|
||||
def _get_main_courses(self) -> str:
|
||||
acc = ""
|
||||
|
||||
for meal in self.meals:
|
||||
if meal.type != _MealType.MAIN:
|
||||
continue
|
||||
|
||||
prefix = self._get_dutch_meal_prefix(meal)
|
||||
|
||||
acc += f"* {prefix}: {meal.name} ({meal.price})\n"
|
||||
|
||||
return acc.strip()
|
||||
|
||||
def _get_cold_meals(self) -> str:
|
||||
acc = ""
|
||||
|
||||
for meal in self.meals:
|
||||
if meal.type == _MealType.COLD:
|
||||
acc += f"* {self._get_dutch_meal_prefix(meal)}: {meal.name} ({meal.price})\n"
|
||||
|
||||
return acc.strip()
|
||||
|
||||
def _closed_embed(self, embed: discord.Embed) -> discord.Embed:
|
||||
embed.colour = ghent_university_yellow()
|
||||
embed.description = "The restaurants are closed today."
|
||||
return embed
|
||||
|
||||
def _regular_embed(self, embed: discord.Embed) -> discord.Embed:
|
||||
embed.add_field(name="🥣 Soep", value=self._get_soups(), inline=False)
|
||||
embed.add_field(name="🍴 Hoofdgerechten", value=self._get_main_courses(), inline=False)
|
||||
embed.add_field(name="❄️Koud", value=self._get_cold_meals(), inline=False)
|
||||
|
||||
vegetables = "\n".join(list(sorted(self.vegetables)))
|
||||
embed.add_field(name="🥦 Groenten", value=vegetables, inline=False)
|
||||
|
||||
return embed
|
||||
|
||||
@overrides
|
||||
def to_embed(self, **kwargs) -> discord.Embed:
|
||||
day_dt: date = cast(date, kwargs.get("day_dt"))
|
||||
weekday = int_to_weekday(day_dt.weekday())
|
||||
formatted_date = f"{leading('0', str(day_dt.day))}/{leading('0', str(day_dt.month))}/{day_dt.year}"
|
||||
|
||||
embed = discord.Embed(title=f"Menu - {weekday} {formatted_date}", colour=ghent_university_blue())
|
||||
|
||||
embed = self._regular_embed(embed) if self.open else self._closed_embed(embed)
|
||||
|
||||
if self.message:
|
||||
embed.add_field(name="📣 Extra Mededeling", value=self.message, inline=False)
|
||||
|
||||
return embed
|
||||
|
||||
|
||||
def no_menu_found(day_dt: date) -> discord.Embed:
|
||||
"""Return a different embed if no menu could be found"""
|
||||
embed = discord.Embed(title="Menu", colour=discord.Colour.red())
|
||||
embed.description = f"Unable to retrieve menu for {day_dt.strftime('%d/%m/%Y')}."
|
||||
return embed
|
||||
|
|
@ -15,6 +15,7 @@ import settings
|
|||
from database.crud import ufora_announcements as crud
|
||||
from database.schemas.relational import UforaCourse
|
||||
from didier.data.embeds.base import EmbedBaseModel
|
||||
from didier.utils.discord.colours import ghent_university_blue
|
||||
from didier.utils.types.datetime import int_to_weekday
|
||||
from didier.utils.types.string import leading
|
||||
|
||||
|
|
@ -47,9 +48,9 @@ class UforaNotification(EmbedBaseModel):
|
|||
self.published_dt = self._published_datetime()
|
||||
self._published = self._get_published()
|
||||
|
||||
def to_embed(self, **kwargs: dict) -> discord.Embed:
|
||||
def to_embed(self, **kwargs) -> discord.Embed:
|
||||
"""Turn the notification into an embed"""
|
||||
embed = discord.Embed(colour=discord.Colour.from_rgb(30, 100, 200))
|
||||
embed = discord.Embed(title=self._title, colour=ghent_university_blue())
|
||||
|
||||
embed.set_author(name=self.course.name)
|
||||
embed.title = self._title
|
||||
|
|
|
|||
|
|
@ -46,9 +46,8 @@ class Definition(EmbedPydantic):
|
|||
return string_utils.abbreviate(field, max_length=Limits.EMBED_FIELD_VALUE_LENGTH)
|
||||
|
||||
@overrides
|
||||
def to_embed(self, **kwargs: dict) -> discord.Embed:
|
||||
embed = discord.Embed(colour=colours.urban_dictionary_green())
|
||||
embed.set_author(name="Urban Dictionary")
|
||||
def to_embed(self, **kwargs) -> discord.Embed:
|
||||
embed = discord.Embed(title="Urban Dictionary", colour=colours.urban_dictionary_green())
|
||||
|
||||
embed.add_field(name="Term", value=self.word, inline=True)
|
||||
embed.add_field(name="Author", value=self.author, inline=True)
|
||||
|
|
|
|||
|
|
@ -126,7 +126,7 @@ class WordleErrorEmbed(EmbedBaseModel):
|
|||
message: str
|
||||
|
||||
@overrides
|
||||
def to_embed(self, **kwargs: dict) -> discord.Embed:
|
||||
def to_embed(self, **kwargs) -> discord.Embed:
|
||||
embed = discord.Embed(colour=discord.Colour.red(), title="Wordle")
|
||||
embed.description = self.message
|
||||
embed.set_footer(text=footer())
|
||||
|
|
|
|||
|
|
@ -0,0 +1,31 @@
|
|||
from discord import app_commands
|
||||
|
||||
__all__ = ["autocomplete_day"]
|
||||
|
||||
|
||||
def autocomplete_day(argument: str) -> list[app_commands.Choice]:
|
||||
"""Autocompletion for day-arguments
|
||||
|
||||
This supports relative offsets ("tomorrow") as well as weekdays
|
||||
"""
|
||||
argument = argument.lower()
|
||||
values = [
|
||||
"Tomorrow",
|
||||
"Monday",
|
||||
"Tuesday",
|
||||
"Wednesday",
|
||||
"Thursday",
|
||||
"Friday",
|
||||
"Sunday",
|
||||
"Morgen",
|
||||
"Overmorgen",
|
||||
"Maandag",
|
||||
"Dinsdag",
|
||||
"Woensdag",
|
||||
"Donderdag",
|
||||
"Vrijdag",
|
||||
"Zaterdag",
|
||||
"Zondag",
|
||||
]
|
||||
|
||||
return [app_commands.Choice(name=value, value=value.lower()) for value in values if argument in value.lower()]
|
||||
|
|
@ -1,6 +1,14 @@
|
|||
import discord
|
||||
|
||||
__all__ = ["urban_dictionary_green"]
|
||||
__all__ = ["ghent_university_blue", "ghent_university_yellow", "urban_dictionary_green"]
|
||||
|
||||
|
||||
def ghent_university_blue() -> discord.Colour:
|
||||
return discord.Colour.from_rgb(30, 100, 200)
|
||||
|
||||
|
||||
def ghent_university_yellow() -> discord.Colour:
|
||||
return discord.Colour.from_rgb(255, 210, 0)
|
||||
|
||||
|
||||
def urban_dictionary_green() -> discord.Colour:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,66 @@
|
|||
import contextlib
|
||||
import datetime
|
||||
from datetime import date, timedelta
|
||||
from typing import Optional, Union
|
||||
|
||||
import discord
|
||||
from discord import app_commands
|
||||
from discord.ext.commands import ArgumentParsingError
|
||||
from overrides import overrides
|
||||
|
||||
from didier.utils.discord.autocompletion.time import autocomplete_day
|
||||
from didier.utils.types.datetime import (
|
||||
forward_to_next_weekday,
|
||||
parse_dm_string,
|
||||
str_to_weekday,
|
||||
)
|
||||
|
||||
__all__ = ["date_converter"]
|
||||
|
||||
|
||||
def date_converter(argument: Optional[str]) -> date:
|
||||
"""Converter to turn a string into a date"""
|
||||
# Store original argument for error message purposes
|
||||
original_argument = argument
|
||||
|
||||
# Default to today
|
||||
if not argument:
|
||||
return date.today()
|
||||
|
||||
argument = argument.lower()
|
||||
|
||||
# Manual offsets
|
||||
if argument in (
|
||||
"tomorrow",
|
||||
"tmrw",
|
||||
"morgen",
|
||||
):
|
||||
return date.today() + timedelta(days=1)
|
||||
|
||||
if argument in ("overmorgen",):
|
||||
return date.today() + timedelta(days=2)
|
||||
|
||||
# Weekdays passed in words
|
||||
with contextlib.suppress(ValueError):
|
||||
weekday = str_to_weekday(argument)
|
||||
return forward_to_next_weekday(date.today(), weekday, allow_today=False)
|
||||
|
||||
# Date strings
|
||||
with contextlib.suppress(ValueError):
|
||||
return parse_dm_string(argument)
|
||||
|
||||
# Unparseable
|
||||
raise ArgumentParsingError(f"Unable to interpret `{original_argument}` as a date.")
|
||||
|
||||
|
||||
class DateTransformer(app_commands.Transformer):
|
||||
"""Application commands transformer for dates"""
|
||||
|
||||
async def autocomplete(
|
||||
self, interaction: discord.Interaction, value: Union[int, float, str]
|
||||
) -> list[app_commands.Choice[Union[int, float, str]]]:
|
||||
return autocomplete_day(str(value))
|
||||
|
||||
@overrides
|
||||
async def transform(self, interaction: discord.Interaction, value: str) -> datetime.date:
|
||||
return date_converter(value)
|
||||
|
|
@ -2,7 +2,7 @@ import logging
|
|||
from contextlib import asynccontextmanager
|
||||
from typing import AsyncGenerator
|
||||
|
||||
from aiohttp import ClientResponse, ClientSession
|
||||
from aiohttp import ClientResponse, ClientSession, ContentTypeError
|
||||
|
||||
from didier.exceptions.http_exception import HTTPException
|
||||
|
||||
|
|
@ -18,13 +18,19 @@ def request_successful(response: ClientResponse) -> bool:
|
|||
|
||||
|
||||
@asynccontextmanager
|
||||
async def ensure_get(http_session: ClientSession, endpoint: str) -> AsyncGenerator[dict, None]:
|
||||
async def ensure_get(
|
||||
http_session: ClientSession, endpoint: str, *, log_exceptions: bool = True
|
||||
) -> AsyncGenerator[dict, None]:
|
||||
"""Context manager that automatically raises an exception if a GET-request fails"""
|
||||
async with http_session.get(endpoint) as response:
|
||||
try:
|
||||
content = await response.json()
|
||||
except ContentTypeError:
|
||||
content = await response.text()
|
||||
|
||||
if not request_successful(response):
|
||||
logger.error(
|
||||
"Failed HTTP request to %s (status %s)\nResponse: %s", endpoint, response.status, await response.json()
|
||||
)
|
||||
if log_exceptions:
|
||||
logger.error("Failed HTTP request to %s (status %s)\nResponse: %s", endpoint, response.status, content)
|
||||
|
||||
raise HTTPException(response.status)
|
||||
|
||||
|
|
@ -33,18 +39,29 @@ async def ensure_get(http_session: ClientSession, endpoint: str) -> AsyncGenerat
|
|||
|
||||
@asynccontextmanager
|
||||
async def ensure_post(
|
||||
http_session: ClientSession, endpoint: str, payload: dict, *, expect_return: bool = True
|
||||
http_session: ClientSession,
|
||||
endpoint: str,
|
||||
payload: dict,
|
||||
*,
|
||||
log_exceptions: bool = True,
|
||||
expect_return: bool = True
|
||||
) -> AsyncGenerator[dict, None]:
|
||||
"""Context manager that automatically raises an exception if a POST-request fails"""
|
||||
async with http_session.post(endpoint, data=payload) as response:
|
||||
if not request_successful(response):
|
||||
logger.error(
|
||||
"Failed HTTP request to %s (status %s)\nPayload: %s\nResponse: %s",
|
||||
endpoint,
|
||||
response.status,
|
||||
payload,
|
||||
await response.json(),
|
||||
)
|
||||
try:
|
||||
content = await response.json()
|
||||
except ContentTypeError:
|
||||
content = await response.text()
|
||||
|
||||
if log_exceptions:
|
||||
logger.error(
|
||||
"Failed HTTP request to %s (status %s)\nPayload: %s\nResponse: %s",
|
||||
endpoint,
|
||||
response.status,
|
||||
payload,
|
||||
content,
|
||||
)
|
||||
|
||||
raise HTTPException(response.status)
|
||||
|
||||
|
|
|
|||
|
|
@ -1,18 +1,91 @@
|
|||
import datetime
|
||||
import re
|
||||
import zoneinfo
|
||||
from typing import TypeVar, Union
|
||||
|
||||
__all__ = ["LOCAL_TIMEZONE", "int_to_weekday", "str_to_date", "tz_aware_now"]
|
||||
__all__ = [
|
||||
"LOCAL_TIMEZONE",
|
||||
"forward_to_next_weekday",
|
||||
"int_to_weekday",
|
||||
"parse_dm_string",
|
||||
"str_to_date",
|
||||
"str_to_month",
|
||||
"str_to_weekday",
|
||||
"tz_aware_now",
|
||||
]
|
||||
|
||||
from typing import Union
|
||||
DateType = TypeVar("DateType", datetime.date, datetime.datetime)
|
||||
|
||||
LOCAL_TIMEZONE = zoneinfo.ZoneInfo("Europe/Brussels")
|
||||
|
||||
|
||||
def forward_to_next_weekday(day_dt: DateType, target_weekday: int, *, allow_today: bool = False) -> DateType:
|
||||
"""Forward a date to the next occurence of a weekday"""
|
||||
if not 0 <= target_weekday <= 6:
|
||||
raise ValueError
|
||||
|
||||
# Skip at least one day
|
||||
if not allow_today:
|
||||
day_dt += datetime.timedelta(days=1)
|
||||
|
||||
while day_dt.weekday() != target_weekday:
|
||||
day_dt += datetime.timedelta(days=1)
|
||||
|
||||
return day_dt
|
||||
|
||||
|
||||
def int_to_weekday(number: int) -> str: # pragma: no cover # it's useless to write a test for this
|
||||
"""Get the Dutch name of a weekday from the number"""
|
||||
return ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"][number]
|
||||
|
||||
|
||||
def parse_dm_string(argument: str) -> datetime.date:
|
||||
"""Parse a string to [day]/[month]
|
||||
|
||||
The year is set to the current year by default, as this can be changed easily.
|
||||
|
||||
This supports:
|
||||
- DD/MM
|
||||
- DD (month defaults to current)
|
||||
- DD [Dutch Month, possibly abbreviated]
|
||||
- DD [English Month, possibly abbreviated]
|
||||
- [Dutch Month, possibly abbreviated] DD
|
||||
- [English Month, possibly abbreviated] DD
|
||||
"""
|
||||
argument = argument.lower()
|
||||
today = datetime.date.today()
|
||||
|
||||
# DD/MM
|
||||
if "/" in argument:
|
||||
spl = argument.split("/")
|
||||
if len(spl) != 2:
|
||||
raise ValueError
|
||||
|
||||
return datetime.date(day=int(spl[0]), month=int(spl[1]), year=today.year)
|
||||
|
||||
# Try to interpret text
|
||||
spl = argument.split(" ")
|
||||
if len(spl) != 2:
|
||||
raise ValueError
|
||||
|
||||
# Day Month
|
||||
match = re.search(r"\d+", spl[0]).group()
|
||||
if match is not None:
|
||||
day = int(match)
|
||||
month = str_to_month(spl[1])
|
||||
return datetime.date(day=day, month=month, year=today.year)
|
||||
|
||||
# Month Day
|
||||
match = re.search(r"\d+", spl[0]).group()
|
||||
if match is not None:
|
||||
day = int(match)
|
||||
month = str_to_month(spl[0])
|
||||
return datetime.date(day=day, month=month, year=today.year)
|
||||
|
||||
# Unparseable
|
||||
raise ValueError
|
||||
|
||||
|
||||
def str_to_date(date_str: str, formats: Union[list[str], str] = "%d/%m/%Y") -> datetime.date:
|
||||
"""Turn a string into a DD/MM/YYYY date"""
|
||||
# Allow passing multiple formats in a list
|
||||
|
|
@ -28,6 +101,76 @@ def str_to_date(date_str: str, formats: Union[list[str], str] = "%d/%m/%Y") -> d
|
|||
raise ValueError
|
||||
|
||||
|
||||
def str_to_month(argument: str) -> int:
|
||||
"""Turn a string int oa month, bilingual"""
|
||||
argument = argument.lower()
|
||||
|
||||
month_dict = {
|
||||
# English
|
||||
"january": 1,
|
||||
"february": 2,
|
||||
"march": 3,
|
||||
"april": 4,
|
||||
"may": 5,
|
||||
"june": 6,
|
||||
"july": 7,
|
||||
# August is a prefix of Augustus so it is skipped
|
||||
"september": 9,
|
||||
"october": 10,
|
||||
"november": 11,
|
||||
"december": 12,
|
||||
# Dutch
|
||||
"januari": 1,
|
||||
"februari": 2,
|
||||
"maart": 3,
|
||||
# April is the same in English so it is skipped
|
||||
"mei": 5,
|
||||
"juni": 6,
|
||||
"juli": 7,
|
||||
"augustus": 8,
|
||||
# September is the same in English so it is skipped
|
||||
"oktober": 10,
|
||||
# November is the same in English so it is skipped
|
||||
# December is the same in English so it is skipped
|
||||
}
|
||||
|
||||
for key, value in month_dict.items():
|
||||
if key.startswith(argument):
|
||||
return value
|
||||
|
||||
raise ValueError
|
||||
|
||||
|
||||
def str_to_weekday(argument: str) -> int:
|
||||
"""Turn a string into a weekday, bilingual"""
|
||||
argument = argument.lower()
|
||||
|
||||
weekday_dict = {
|
||||
# English
|
||||
"monday": 0,
|
||||
"tuesday": 1,
|
||||
"wednesday": 2,
|
||||
"thursday": 3,
|
||||
"friday": 4,
|
||||
"saturday": 5,
|
||||
"sunday": 6,
|
||||
# Dutch
|
||||
"maandag": 0,
|
||||
"dinsdag": 1,
|
||||
"woensdag": 2,
|
||||
"donderdag": 3,
|
||||
"vrijdag": 4,
|
||||
"zaterdag": 5,
|
||||
"zondag": 6,
|
||||
}
|
||||
|
||||
for key, value in weekday_dict.items():
|
||||
if key.startswith(argument):
|
||||
return value
|
||||
|
||||
raise ValueError
|
||||
|
||||
|
||||
def tz_aware_now() -> datetime.datetime:
|
||||
"""Get the current date & time, but timezone-aware"""
|
||||
return datetime.datetime.utcnow().replace(tzinfo=datetime.timezone.utc).astimezone(LOCAL_TIMEZONE)
|
||||
|
|
|
|||
Loading…
Reference in New Issue