PK!'Gcrunchy_api/__init__.pyimport logging import json import random from string import Template import typing from urllib.parse import urlencode from urllib.request import urlopen from crunchy_api.types import Field, ObjectType, MediaType, SortMode __version__ = "0.3.0" """ Api documentation https://github.com/CloudMax94/crunchyroll-api/wiki/Api """ API_VERSION = "0" URL = Template(f"https://api.crunchyroll.com/$method.{API_VERSION}.json") VERSION = "1.1.21.0" DEVICE = "com.crunchyroll.windows.desktop" LOGGER = logging.getLogger(__name__) LOGGER.setLevel(logging.INFO) def _make_request(method: str, payload: dict) -> dict: endpoint = URL.substitute(method=method) data = urlencode(payload).encode("utf-8") LOGGER.info("endpoint: %s\ndata: %s", endpoint, data) request = urlopen(endpoint, data) return json.loads(request.read().decode("utf-8")) def _refresh_session( device_id, device_type, access_token, version, locale, auth=None ) -> str: payload = { "device_id": device_id, "device_type": device_type, "access_token": access_token, "version": version, "locale": locale, } if auth: payload.update(auth=auth) result = _make_request("start_session", payload) return result["data"]["session_id"] # Attempt to refresh the session and try again if the request fails def retry_on_error(func): def retry_func(*args, **kwargs): attempt = func(*args, **kwargs) if attempt["error"]: self = args[0] # This feels clever in the worst kind of way self._session_id = _refresh_session( self._device_id, # pylint: disable=protected-access DEVICE, self._token, # pylint: disable=protected-access VERSION, self.locale, self._auth, # pylint: disable=protected-access ) attempt = func(*args, **kwargs) if attempt["error"]: # Couldn't restart the session, kill everything, try authentication from the start self._session_id = None self._auth = None return attempt return retry_func SORT = {SortMode.ASC: "asc", SortMode.DESC: "desc"} INFO = { ObjectType.MEDIA: "media_id", ObjectType.COLLECTION: "collection_id", ObjectType.SERIES: "series_id", } MEDIA = { MediaType.ANIME: "anime", MediaType.DRAMA: "drama", MediaType.ANIME_DRAMA: "anime|drama", } class CrunchyrollApi: def _request(self, method, args): payload = args.copy() payload.update(version=VERSION, locale=self.locale, session_id=self._session) return _make_request(method, payload) def __init__(self, username, password, token, locale="enUS") -> None: self._session_id = None self._auth = None self._username = username self._password = password self._token = token char_set = "0123456789abcdefghijklmnopqrstuvwxyz0123456789" self._device_id = "".join( [ "".join(random.sample(char_set, 8)), "-KODI-", "".join(random.sample(char_set, 4)), "-", "".join(random.sample(char_set, 4)), "-", "".join(random.sample(char_set, 12)), ] ) self.locale = locale self.login(username, password) @property def _session(self): if not self._session_id: self._session_id = _refresh_session( self._device_id, DEVICE, self._token, VERSION, self.locale, self._auth ) return self._session_id @retry_on_error def add_to_queue(self, series_id: int) -> dict: raise NotImplementedError("Haven't gotten to this yet") @retry_on_error def batch(self) -> dict: raise NotImplementedError("Haven't gotten to this yet") @retry_on_error def categories(self) -> dict: raise NotImplementedError("Haven't gotten to this yet") @retry_on_error def info(self, object_type: ObjectType, object_id: int) -> dict: payload = {INFO[object_type]: object_id} return self._request("info", payload) @retry_on_error def list_locales(self) -> dict: raise NotImplementedError("Haven't gotten to this yet") @retry_on_error def list_media( self, object_type: ObjectType, object_id: int, sort: SortMode = SortMode.ASC, offset: int = 0, locale: str = None, ) -> dict: locale = locale or self.locale payload = { INFO[object_type]: object_id, sort: SORT[sort], offset: offset, locale: locale, } return self._request("list_media", payload) @retry_on_error def list_series(self) -> dict: raise NotImplementedError("Haven't gotten to this yet") @retry_on_error def log(self) -> dict: raise NotImplementedError("Haven't gotten to this yet") @retry_on_error def login(self, username, password) -> dict: payload = {"account": username, "password": password} result = self._request("login", payload) self._auth = result["data"]["auth"] return result @retry_on_error def logout(self) -> dict: payload = {"auth": self._auth} result = self._request("logout", payload) self._auth = None return result @retry_on_error def queue( self, media_types: MediaType, fields: typing.Iterable[Field] = tuple() ) -> dict: payload = { "media_types": MEDIA[media_types], "fields": ",".join([Field.to_str(field) for field in fields]), } return self._request("queue", payload) @retry_on_error def recently_watched(self) -> dict: raise NotImplementedError("Haven't gotten to this yet") @retry_on_error def remove_from_queue(self) -> dict: raise NotImplementedError("Haven't gotten to this yet") PK!_Oc crunchy_api/cli.pyimport json try: import click except ImportError: raise ImportError("Please install the optional `cli` extras to use the CLI.") import crunchy_api from crunchy_api import CrunchyrollApi from crunchy_api import types @click.group() @click.option("--username", required=True, envvar="CRUNCHYROLL_USERNAME") @click.option("--password", envvar="CRUNCHYROLL_PASSWORD") @click.option("--token", required=True, envvar="CRUNCHYROLL_TOKEN") @click.option("--locale", default="enUS", envvar="CRUNCHYROLL_LOCALE") @click.option( "--stdin-password", is_flag=True, default=False, help="Read password in from stdin" ) @click.pass_context def main( ctx: click.core.Context, username: str, password: str, token: str, locale: str, stdin_password: bool, ) -> None: if not (password or stdin_password): raise click.UsageError("Must supply one of `password` or `stdin_password`") if stdin_password: password = input() ctx.obj = CrunchyrollApi(username, password, token) @main.command() @click.option( "--object-type", required=True, type=click.Choice(list(crunchy_api.INFO.values())), default="series_id", envvar="CRUNCHYROLL_OBJECT", ) @click.option("--object-id", required=True, type=int, envvar="CRUNCHYROLL_ID") @click.pass_context def info(ctx: click.core.Context, object_type: str, object_id: int) -> None: obj = types.ObjectType.from_str(object_type) api = ctx.obj click.echo(api.info(obj, object_id)) api.logout() @main.command() @click.option( "--object-type", required=True, type=click.Choice(list(crunchy_api.INFO.values())), default="series_id", envvar="CRUNCHYROLL_OBJECT", ) @click.option("--object-id", required=True, type=int, envvar="CRUNCHYROLL_ID") @click.option( "--sort", type=click.Choice(list(crunchy_api.SORT.values())), default="asc", envvar="CRUNCHYROLL_SORT", ) @click.option("--offset", type=int, default=0, envvar="CRUNCHYROLL_OFFSET") @click.option("--locale", type=str, default="enUS", envvar="CRUNCHYROLL_LOCALE") @click.pass_context def list_media( ctx: click.core.Context, object_type: str, object_id: int, sort: str, offset: int, locale: str, ) -> None: obj = types.ObjectType.from_str(object_type) sort = types.SortMode.from_str(sort) api = ctx.obj click.echo(api.list_media(obj, object_id, sort, offset, locale)) api.logout() @main.command() @click.option( "--media-type", type=click.Choice(list(crunchy_api.MEDIA.values())), default="anime", envvar="CRUNCHYROLL_MEDIA", ) @click.pass_context def queue(ctx: click.core.Context, media_type: str) -> None: media = types.MediaType.from_str(media_type) api = ctx.obj click.echo(api.queue(media)) api.logout() PK! crunchy_api/types.pyimport enum class Field(enum.Enum): IMAGE_FULL_URL = enum.auto() IMAGE_FWIDE_URL = enum.auto() IMAGE_FWIDESTAR_URL = enum.auto() IMAGE_HEIGHT = enum.auto() IMAGE_LARGE_URL = enum.auto() IMAGE_MEDIUM_URL = enum.auto() IMAGE_SMALL_URL = enum.auto() IMAGE_THUMB_URL = enum.auto() IMAGE_WIDE_URL = enum.auto() IMAGE_WIDESTAR_URL = enum.auto() IMAGE_WIDTH = enum.auto() MEDIA_AVAILABILITY_NOTES = enum.auto() MEDIA_AVAILABLE = enum.auto() MEDIA_AVAILABLE_TIME = enum.auto() MEDIA_BIF_URL = enum.auto() MEDIA_CLASS = enum.auto() MEDIA_CLIP = enum.auto() MEDIA_COLLECTION_ID = enum.auto() MEDIA_COLLECTION_NAME = enum.auto() MEDIA_CREATED = enum.auto() MEDIA_DESCRIPTION = enum.auto() MEDIA_DURATION = enum.auto() MEDIA_EPISODE_NUMBER = enum.auto() MEDIA_FREE_AVAILABLE = enum.auto() MEDIA_FREE_AVAILABLE_TIME = enum.auto() MEDIA_FREE_UNAVAILABLE_TIME = enum.auto() MEDIA_MEDIA_ID = enum.auto() MEDIA_MEDIA_TYPE = enum.auto() MEDIA_NAME = enum.auto() MEDIA_PLAYHEAD = enum.auto() MEDIA_PREMIUM_AVAILABLE = enum.auto() MEDIA_PREMIUM_AVAILABLE_TIME = enum.auto() MEDIA_PREMIUM_ONLY = enum.auto() MEDIA_PREMIUM_UNAVAILABLE_TIME = enum.auto() MEDIA_SCREENSHOT_IMAGE = enum.auto() MEDIA_SERIES_ID = enum.auto() MEDIA_SERIES_NAME = enum.auto() MEDIA_STREAM_DATA = enum.auto() MEDIA_UNAVAILABLE_TIME = enum.auto() MEDIA_URL = enum.auto() LAST_WATCHED_MEDIA = enum.auto() LAST_WATCHED_MEDIA_PLAYHEAD = enum.auto() MOST_LIKELY_MEDIA = enum.auto() MOST_LIKELY_MEDIA_PLAYHEAD = enum.auto() ORDERING = enum.auto() PLAYHEAD = enum.auto() QUEUE_ENTRY_ID = enum.auto() SERIES = enum.auto() SERIES_CLASS = enum.auto() SERIES_COLLECTION_COUNT = enum.auto() SERIES_DESCRIPTION = enum.auto() SERIES_GENRES = enum.auto() SERIES_IN_QUEUE = enum.auto() SERIES_LANDSCAPE_IMAGE = enum.auto() SERIES_MEDIA_COUNT = enum.auto() SERIES_MEDIA_TYPE = enum.auto() SERIES_NAME = enum.auto() SERIES_PORTRAIT_IMAGE = enum.auto() SERIES_PUBLISHER_NAME = enum.auto() SERIES_RATING = enum.auto() SERIES_SERIES_ID = enum.auto() SERIES_URL = enum.auto() SERIES_YEAR = enum.auto() @staticmethod def from_str(field: str) -> 'Field': return getattr(Field, field.replace('.', '_').upper()) #TODO: Replace this with a map @staticmethod def to_str(field: "Field") -> str: return str(field).replace("_", ".").lower() class MediaType(enum.Enum): ANIME = enum.auto() DRAMA = enum.auto() ANIME_DRAMA = enum.auto() @staticmethod def from_str(media: str) -> 'MediaType': return getattr(MediaType, media.replace('|', '_').upper()) class ObjectType(enum.Enum): MEDIA = enum.auto() COLLECTION = enum.auto() SERIES = enum.auto() @staticmethod def from_str(obj: str) -> 'MediaType': return getattr(ObjectType, obj.upper()[:-3]) class SortMode(enum.Enum): ASC = enum.auto() DESC = enum.auto() @staticmethod def from_str(sort: str) -> 'MediaType': return getattr(SortMode, sort.upper()) PK!HZ3,0,crunchy_api-0.3.0.dist-info/entry_points.txtN+I/N.,()J.*KΨz9Vy\\PK!z,..#crunchy_api-0.3.0.dist-info/LICENSEMIT License Copyright (c) 2018 David Buckley Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. PK!H_zTT!crunchy_api-0.3.0.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]n0H*J>mlcAPK!H|;We$crunchy_api-0.3.0.dist-info/METADATARN0}W&C1TF`" ({ְw0Eyk=瞞;^$ )]pț($U:j*"..YVE!\aQh;at1|&a|:-Z{[ HϪ5Mx˚i"v* T,utE I#RYsh=; {Jr#C>W,Fa;44EY8}9Zgv2ieLDQ(LB <.ųŌAzF/ 0W'UC)(Ag8`G {E&cԞj!xD61{MَH7a >-!?PK!H)"crunchy_api-0.3.0.dist-info/RECORD};0~lCPVdžA ` 4pq/^8iFw"h dk&O{Hihf>SYQJ]8 hW5z`UƚGci=#h<m]iI)@;-WCiNEᕝ]K9Y;ApB.gL!%wi'PF_wBg ؞~(kN4$X()brѴP/x=$}BFT+ARegY+N:hG3Ydu6q~H(]IQ!$ze{R sP>^\96ʆHaRӇjfI*r-/4Qh }PK!'Gcrunchy_api/__init__.pyPK!_Oc crunchy_api/cli.pyPK! "crunchy_api/types.pyPK!HZ3,0,/crunchy_api-0.3.0.dist-info/entry_points.txtPK!z,..#(0crunchy_api-0.3.0.dist-info/LICENSEPK!H_zTT!4crunchy_api-0.3.0.dist-info/WHEELPK!H|;We$*5crunchy_api-0.3.0.dist-info/METADATAPK!H)"6crunchy_api-0.3.0.dist-info/RECORDPKc8