major cleanup

- `routers`: `admin`, `days`, `general`, `user` -> `admin`, `images`
- `core.depends`: cleanup
- `core.*_helpers`: joined in `core.helpers`
This commit is contained in:
Jörn-Michael Miehe 2023-09-12 13:50:02 +00:00
parent 63d88c3a09
commit 3316bf2822
11 changed files with 132 additions and 203 deletions

View file

@ -7,8 +7,7 @@ from PIL import Image, ImageFont
from .advent_image import _XY, AdventImage
from .calendar_config import CalendarConfig, get_calendar_config
from .config import Config, get_config
from .image_helpers import list_images_auto, load_image
from .sequence_helpers import Random, set_len, shuffle
from .helpers import Random, list_images_auto, load_image, set_len
from .webdav import WebDAV
@ -19,7 +18,7 @@ async def get_days(
Alle Tage, für die es ein Türchen gibt
"""
return list(set(door.day for door in cal_cfg.doors))
return sorted(set(door.day for door in cal_cfg.doors))
async def get_day_parts(
@ -47,18 +46,7 @@ async def get_day_parts(
return result
async def shuffle_images_auto(
images: list[str] = Depends(list_images_auto),
) -> list[str]:
"""
Bilder: Reihenfolge zufällig bestimmen
"""
ls = set_len(images, 24)
return await shuffle(ls)
async def get_part_for_day(
async def get_day_part(
day: int,
parts: dict[int, str] = Depends(get_day_parts),
) -> str:
@ -69,38 +57,43 @@ async def get_part_for_day(
return parts[day]
async def get_random_for_day(
day: int,
) -> Random:
async def get_auto_image_names(
days: list[int] = Depends(get_days),
images: list[str] = Depends(list_images_auto),
) -> dict[int, str]:
"""
Tagesabhängige Zufallszahlen
Bilder: Reihenfolge zufällig bestimmen
"""
return await Random.get(day)
rnd = await Random.get()
ls = set_len(images, len(days))
return dict(zip(days, rnd.shuffled(ls)))
async def gen_auto_image_for_day(
async def gen_day_auto_image(
day: int,
auto_images: list[str] = Depends(shuffle_images_auto),
cfg: Config = Depends(get_config),
rnd: Random = Depends(get_random_for_day),
part: str = Depends(get_part_for_day),
auto_image_names: list[str] = Depends(get_auto_image_names),
day_part: str = Depends(get_day_part),
) -> Image.Image:
"""
Automatisch generiertes Bild erstellen
"""
# Datei existiert garantiert!
img = await load_image(auto_images[day])
img = await load_image(auto_image_names[day])
image = await AdventImage.from_img(img)
rnd = await Random.get(day)
font = ImageFont.truetype(
font=BytesIO(await WebDAV.read_bytes(f"files/{cfg.server.font}")),
size=50,
)
# Buchstaben verstecken
for letter in part:
for letter in day_part:
await image.hide_text(
xy=cast(_XY, tuple(rnd.choices(range(30, 470), k=2))),
text=letter,
@ -110,12 +103,11 @@ async def gen_auto_image_for_day(
return image.img
async def get_image_for_day(
async def get_day_image(
day: int,
auto_images: list[str] = Depends(shuffle_images_auto),
cfg: Config = Depends(get_config),
rnd: Random = Depends(get_random_for_day),
part: str = Depends(get_part_for_day),
auto_image_names: list[str] = Depends(get_auto_image_names),
day_part: str = Depends(get_day_part),
) -> Image.Image:
"""
Bild für einen Tag abrufen
@ -131,6 +123,6 @@ async def get_image_for_day(
except RuntimeError:
# Erstelle automatisch generiertes Bild
return await gen_auto_image_for_day(
day=day, auto_images=auto_images, cfg=cfg, rnd=rnd, part=part
return await gen_day_auto_image(
day=day, cfg=cfg, auto_image_names=auto_image_names, day_part=day_part
)

View file

@ -1,11 +1,35 @@
import itertools
import random
import re
from io import BytesIO
from typing import Any, Self, Sequence, TypeVar
from fastapi.responses import StreamingResponse
from PIL import Image
from .config import get_config
from .webdav import WebDAV
T = TypeVar("T")
class Random(random.Random):
@classmethod
async def get(cls, bonus_salt: Any = "") -> Self:
cfg = await get_config()
return cls(f"{cfg.puzzle.solution}{cfg.puzzle.random_seed}{bonus_salt}")
def shuffled(self, population: Sequence[T]) -> Sequence[T]:
return self.sample(population, k=len(population))
def set_len(seq: Sequence[T], len: int) -> Sequence[T]:
# `seq` unendlich wiederholen
infinite = itertools.cycle(seq)
# Die ersten `length` einträge nehmen
return list(itertools.islice(infinite, len))
async def list_images_auto() -> list[str]:
"""

View file

@ -1,33 +0,0 @@
import itertools
import random
from typing import Any, Self, Sequence, TypeVar
from .config import get_config
T = TypeVar("T")
class Random(random.Random):
@classmethod
async def get(cls, bonus_salt: Any = "") -> Self:
cfg = await get_config()
return cls(f"{cfg.puzzle.solution}{cfg.puzzle.random_seed}{bonus_salt}")
def shuffled(self, population: Sequence[T]) -> Sequence[T]:
return self.sample(population, k=len(population))
async def shuffle(seq: Sequence, rnd: random.Random | None = None) -> list:
# Zufallsgenerator
rnd = rnd or await Random.get()
# Elemente mischen
return rnd.sample(seq, len(seq))
def set_len(seq: Sequence, length: int) -> list:
# `seq` unendlich wiederholen
infinite = itertools.cycle(seq)
# Die ersten `length` einträge nehmen
return list(itertools.islice(infinite, length))

View file

@ -1,10 +1,8 @@
from fastapi import APIRouter
from . import admin, days, general, user
from . import admin, images
router = APIRouter(prefix="/api")
router.include_router(admin.router)
router.include_router(days.router)
router.include_router(general.router)
router.include_router(user.router)
router.include_router(images.router)

View file

@ -50,7 +50,7 @@ async def user_visible_doors() -> int:
return 0
async def user_can_view_door(
async def user_can_view_day(
day: int,
) -> bool:
"""

View file

@ -3,9 +3,9 @@ from datetime import date
from fastapi import APIRouter, Depends
from pydantic import BaseModel
from ..core import depends
from ..core.calendar_config import CalendarConfig, get_calendar_config
from ..core.calendar_config import CalendarConfig, DoorsSaved, get_calendar_config
from ..core.config import Config, get_config
from ..core.depends import get_day_parts
from ..core.settings import SETTINGS
from ._security import require_admin, user_is_admin
@ -61,7 +61,7 @@ async def get_config_model(
_: None = Depends(require_admin),
cfg: Config = Depends(get_config),
cal_cfg: CalendarConfig = Depends(get_calendar_config),
day_parts: dict[int, str] = Depends(depends.get_day_parts),
day_parts: dict[int, str] = Depends(get_day_parts),
) -> ConfigModel:
return ConfigModel.model_validate(
{
@ -92,3 +92,31 @@ async def get_config_model(
},
}
)
@router.get("/doors")
async def get_doors(
cal_cfg: CalendarConfig = Depends(get_calendar_config),
) -> DoorsSaved:
"""
Türchen lesen
"""
return cal_cfg.doors
@router.put("/doors")
async def put_doors(
doors: DoorsSaved,
cfg: Config = Depends(get_config),
cal_cfg: CalendarConfig = Depends(get_calendar_config),
) -> None:
"""
Türchen ändern
"""
cal_cfg.doors = sorted(
doors,
key=lambda door: door.day,
)
await cal_cfg.change(cfg)

View file

@ -1,68 +0,0 @@
from datetime import date
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.responses import StreamingResponse
from PIL import Image
from ..core import depends
from ..core.config import get_config
from ..core.image_helpers import api_return_image
from ._security import user_can_view_door, user_is_admin, user_visible_doors
router = APIRouter(prefix="/days", tags=["days"])
@router.on_event("startup")
async def startup() -> None:
cfg = await get_config()
print(cfg.puzzle.solution)
@router.get("/date")
async def get_date() -> str:
"""
Aktuelles Server-Datum
"""
return date.today().isoformat()
@router.get("/visible_days")
async def get_visible_days(
visible_doors: int = Depends(user_visible_doors),
) -> int:
"""
Sichtbare Türchen
"""
return visible_doors
@router.get("/part/{day}")
async def get_part_for_day(
part: str = Depends(depends.get_part_for_day),
) -> str:
"""
Heutiger Lösungsteil
"""
return part
@router.get(
"/image/{day}",
response_class=StreamingResponse,
)
async def get_image_for_day(
image: Image.Image = Depends(depends.get_image_for_day),
can_view: bool = Depends(user_can_view_door),
is_admin: bool = Depends(user_is_admin),
) -> StreamingResponse:
"""
Bild für einen Tag erstellen
"""
if not (can_view or is_admin):
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Wie unhöflich!!!")
return await api_return_image(image)

View file

@ -1,50 +0,0 @@
from fastapi import APIRouter, Depends
from fastapi.responses import StreamingResponse
from ..core.calendar_config import CalendarConfig, DoorsSaved, get_calendar_config
from ..core.config import Config, get_config
from ..core.image_helpers import api_return_image, load_image
router = APIRouter(prefix="/general", tags=["general"])
@router.get(
"/background",
response_class=StreamingResponse,
)
async def get_image_for_day(
cal_cfg: CalendarConfig = Depends(get_calendar_config),
) -> StreamingResponse:
"""
Hintergrundbild laden
"""
return await api_return_image(await load_image(f"files/{cal_cfg.background}"))
@router.get("/doors")
async def get_doors(
cal_cfg: CalendarConfig = Depends(get_calendar_config),
) -> DoorsSaved:
"""
Türchen lesen
"""
return cal_cfg.doors
@router.put("/doors")
async def put_doors(
doors: DoorsSaved,
cfg: Config = Depends(get_config),
cal_cfg: CalendarConfig = Depends(get_calendar_config),
) -> None:
"""
Türchen setzen
"""
cal_cfg.doors = sorted(
doors,
key=lambda door: door.day,
)
await cal_cfg.change(cfg)

View file

@ -0,0 +1,50 @@
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.responses import StreamingResponse
from PIL import Image
from ..core.calendar_config import CalendarConfig, get_calendar_config
from ..core.config import get_config
from ..core.depends import get_day_image
from ..core.helpers import api_return_image, load_image
from ._security import user_can_view_day, user_is_admin
router = APIRouter(prefix="/images", tags=["images"])
@router.on_event("startup")
async def startup() -> None:
cfg = await get_config()
print(cfg.puzzle.solution)
@router.get(
"/background",
response_class=StreamingResponse,
)
async def get_background(
cal_cfg: CalendarConfig = Depends(get_calendar_config),
) -> StreamingResponse:
"""
Hintergrundbild laden
"""
return await api_return_image(await load_image(f"files/{cal_cfg.background}"))
@router.get(
"/{day}",
response_class=StreamingResponse,
)
async def get_image_for_day(
image: Image.Image = Depends(get_day_image),
can_view: bool = Depends(user_can_view_day),
is_admin: bool = Depends(user_is_admin),
) -> StreamingResponse:
"""
Bild für einen Tag erstellen
"""
if not (can_view or is_admin):
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Wie unhöflich!!!")
return await api_return_image(image)

View file

@ -1,12 +0,0 @@
from fastapi import APIRouter, Depends
from ._security import require_admin
router = APIRouter(prefix="/user", tags=["user"])
@router.get("/admin")
def check_admin(
_: None = Depends(require_admin),
) -> bool:
return True