diff --git a/README.md b/README.md index 4fe8921..07dedc5 100644 --- a/README.md +++ b/README.md @@ -10,8 +10,6 @@ https://anilist.abstractumbra.dev/AbstractUmbra/OtherUmbra/etc... By default this will compare entries in the "Planning" category. You can add a `status` query parameter to further refine what category you wish to see! -You can also pass one (or more) `exclude` query parameters to exclude a column. - The `?status=` parameter accepts the following values:- ``` planning (the default) @@ -22,14 +20,11 @@ paused repeating ``` -The `?exclude=` parameter accepts the following values:- -``` -romaji -english -native -``` - - ## Running your own -The provided docker-compose file should work on it's own, otherwise just clone the repository and install the necessary dependencies and run the `main.py` with a Python version >= 3.11 +The provided docker-compose file should work on its own. + +To run in development mode, use +``` +litestar --app anilist-cmp:app run --debug --reload +``` diff --git a/anilist-cmp/__init__.py b/anilist-cmp/__init__.py index 9335e50..8f58356 100644 --- a/anilist-cmp/__init__.py +++ b/anilist-cmp/__init__.py @@ -1,13 +1,16 @@ from __future__ import annotations +import pathlib +from collections import ChainMap from enum import Enum -from functools import reduce -from operator import and_, or_ from typing import TYPE_CHECKING import httpx from litestar import Litestar, MediaType, Response, get, status_codes +from litestar.contrib.jinja import JinjaTemplateEngine from litestar.middleware.rate_limit import RateLimitConfig +from litestar.response import Template +from litestar.template import TemplateConfig if TYPE_CHECKING: from collections.abc import Sequence @@ -38,47 +41,8 @@ }} """ -OPENGRAPH_HEAD = """ - - - - - Hi - - - - - - -""" - -HEADINGS = ["romaji", "english", "native"] - -TABLE = """ - - - - -{included} - - - - -{{body}} - -
Media IDURL
-""" - -ROW = """ - -{{id}} -{included} -Anilist - -""" - -class NoPlanningData(ValueError): +class EmptyAnimeList(ValueError): def __init__(self, user: int, *args: object) -> None: self.user = user super().__init__(*args) @@ -93,13 +57,6 @@ class Status(Enum): repeating = "REPEATING" -def format_entries_as_table(entries: dict[int, InnerMediaEntry], excluded: list[str]) -> str: - included = "\n".join(f"{{title[{h}]}}" for h in HEADINGS if h not in excluded) - rows = [ROW.format(included=included).format_map(entry) for entry in entries.values()] - formatted_headings = "\n".join(f"{h.title()}" for h in HEADINGS if h not in excluded) - return TABLE.format(included=formatted_headings).format(body="\n".join(rows)) - - async def _fetch_user_entries(*usernames: str, status: Status) -> AnilistResponse | AnilistErrorResponse: parameters = ", ".join(f"$username{n}: String" for n in range(len(usernames))) subqueries = "".join(USER_QUERY.format(number=n) for n in range(len(usernames))) @@ -117,17 +74,17 @@ def _restructure_entries(entries: list[MediaEntry]) -> dict[int, InnerMediaEntry return {entry["media"]["id"]: entry["media"] for entry in entries} -def _get_common_planning(data: AnilistResponse) -> dict[int, InnerMediaEntry]: +def _get_common_anime(data: AnilistResponse) -> dict[int, InnerMediaEntry]: media_entries: list[dict[int, InnerMediaEntry]] = [] for index, item in enumerate(data["data"].values()): if not item or not item["lists"]: - raise NoPlanningData(index) + raise EmptyAnimeList(index) media_entries.append(_restructure_entries(item["lists"][0]["entries"])) - all_anime: dict[int, InnerMediaEntry] = reduce(or_, media_entries) - common_anime: set[int] = reduce(and_, (d.keys() for d in media_entries)) + all_anime = ChainMap(*media_entries) + common_anime = set(all_anime).intersection(*media_entries) return {id_: all_anime[id_] for id_ in common_anime} @@ -167,8 +124,9 @@ async def index() -> Response[str]: @get("/{user_list:path}") -async def get_matches(user_list: str, exclude: list[str] | None = None, status: str = "planning") -> Response[str]: - users = list({user.casefold() for user in user_list.lstrip("/").split("/")}) +async def get_matches(user_list: str, status: str = "planning") -> Response[str] | Template: + usernames = [username for username in user_list.split('/') if username] + users = list({user.lower() for user in usernames}) if len(users) <= 1: return Response( @@ -186,22 +144,11 @@ async def get_matches(user_list: str, exclude: list[str] | None = None, status: ) try: - selected_status = Status[status.casefold()] + selected_status = Status[status.lower()] except KeyError: _statuses = "\n".join(item.name for item in Status) return Response(f"Sorry, your chosen status of {status} is not valid. Please choose from:-\n\n{_statuses}") - excluded = list({ex.casefold() for ex in exclude or []}) - - faulty = [ex for ex in excluded if ex not in HEADINGS] - - if faulty: - return Response( - f"Unknown excluded headings: {_human_join(faulty)}. Supported: {_human_join(HEADINGS)}", - media_type=MediaType.TEXT, - status_code=status_codes.HTTP_400_BAD_REQUEST, - ) - data = await _fetch_user_entries(*users, status=selected_status) if errors := data.get("errors"): @@ -215,26 +162,25 @@ async def get_matches(user_list: str, exclude: list[str] | None = None, status: ) try: - matching_items = _get_common_planning(data) # type: ignore # the type is resolved above. - except NoPlanningData as err: + matching_items = _get_common_anime(data) # type: ignore # the type is resolved above. + except EmptyAnimeList as err: errored_user = users[err.user] return Response( f"Sorry, but {errored_user} has no {selected_status.value.lower()} entries!", media_type=MediaType.TEXT, - status_code=status_codes.HTTP_412_PRECONDITION_FAILED, + status_code=status_codes.HTTP_404_NOT_FOUND, ) if not matching_items: return Response( f"No {selected_status.value.lower()} anime in common :(", media_type=MediaType.TEXT, - status_code=status_codes.HTTP_412_PRECONDITION_FAILED, + status_code=status_codes.HTTP_404_NOT_FOUND, ) - head = OPENGRAPH_HEAD.format(users=_human_join(users), mutual=len(matching_items), status=selected_status.value.title()) - formatted = format_entries_as_table(matching_items, excluded=excluded) - - return Response(head + "\n" + formatted, media_type=MediaType.HTML, status_code=status_codes.HTTP_200_OK) + context = dict(entries=sorted(matching_items.values(), key=lambda entry: entry['id']), status=selected_status, + description=f'Common anime for {_human_join(usernames)}') + return Template(template_name='page.html', context=context) RL_CONFIG = RateLimitConfig( @@ -246,9 +192,9 @@ async def get_matches(user_list: str, exclude: list[str] | None = None, status: rate_limit_reset_header_key="X-Ratelimit-Reset", ) - +template_directory = pathlib.Path(__file__).parent / 'templates' app = Litestar( route_handlers=[index, get_matches], middleware=[RL_CONFIG.middleware], - debug=True, + template_config=TemplateConfig(directory=template_directory, engine=JinjaTemplateEngine), ) diff --git a/anilist-cmp/templates/page.html b/anilist-cmp/templates/page.html new file mode 100644 index 0000000..997c428 --- /dev/null +++ b/anilist-cmp/templates/page.html @@ -0,0 +1,62 @@ + + + + + + Anilist common anime tool + + + + + + + + + + + + + + + + + + + + + + + + {% for entry in entries %} + + + + + + + {% endfor %} + +
Media IDRomajiEnglishNative
{{ entry['id'] }}{{ entry['title']['romaji'] }}{{ entry['title']['english'] }}{{ entry['title']['native'] }}
+ + + diff --git a/pyproject.toml b/pyproject.toml index cc37426..4865ff3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,7 +6,7 @@ authors = ["Alex Nørgaard "] [tool.poetry.dependencies] python = "^3.12" -litestar = { version = "*", extras = ["standard"] } +litestar = { version = "*", extras = ["jinja"] } uvicorn = "*" uvloop = { version = "*", markers = "sys_platform != 'win32'" }