Getting started with the API¶

Installation¶

Via poetry:

poetry add anipy-api

Via poetry (from source):

poetry add "git+https://github.com/sdaqo/anipy-cli.git#subdirectory=api"

Via pip:

pip install anipy-api

Via pip (from source):

pip install "git+https://github.com/sdaqo/anipy-cli.git#subdirectory=api"

Provider
What is a provider?¶

A provider in anipy-api is the building stone that connects us to the anime we

want! A provder may be a external anime site or even your local files (this is
planned, Coming Soon!). Every provider bases on the BaseProvider and has the same
basic functions: get_search, get_episodes, get_info and get_video.

You can check which providers anipy-cli supports here or here (in the providers
drop-down).
Getting a provider instance¶

To get a provider you can use the list_providers function, get_provider to get a
provider by its string representation or just simply import it directly.

from anipy_api.provider import list_providers, get_provider

# List providers
for p in list_providers():
if p.NAME == "gogoanime":
# You have to instantiate the provider to use it.
provider = provider()

# If you know the name of the provider you could also do:
provider = get_provider("gogoanime", base_url_override="https://test.com") #

# You can also import

from anipy_api.provider.providers import GoGoProvider
provider = GoGoProvider()

print(provider.NAME) # -> gogoanime

print(provider.BASE_URL) # -> https://test.com

Searching¶

Searching can be done with the get_search method, you can even filter!
results = provider.get_search("frieren") #

This returns a list of ProviderSearchResult objects. Each contains the identifier

of the anime, the name and a set of language types the anime supports. The language
type set tells you if the anime supports dub/sub format.
Filtering¶

Every provider has a FILTER_CAPS attribute, it specifies which filters the provider
supports.

from anipy_api.provider import Filters, FilterCapabilities, Season

if provider.FILTER_CAPS & ( #

FilterCapabilities.SEASON
| FilterCapabilities.YEAR
| FilterCapabilities.NO_QUERY #

):
filters = Filters( #

year=2023,
season=Season.FALL,
)
fall_2023_anime = provider.get_search("", filters=filters) #

Putting it all together¶

from anipy_api.provider import get_provider

provider = get_provider("gogoanime")
frieren = provider.get_search("frieren")[0]

if provider.FILTER_CAPS & (
FilterCapabilities.SEASON
| FilterCapabilities.YEAR
| FilterCapabilities.NO_QUERY
):
filters = Filters(
year=2023,
season=Season.FALL,
)
fall_2023_anime = provider.get_search("", filters=filters)

if frieren in fall_2023_anime:
print("Frieren is an fall 2023 anime!")

Thats about it here¶

But... get_episodes, get_info and get_video were not covered!

Those are covered in the second example page 2. Anime. The next page describes a
wrapper for the search results that represents an anime, you can still use the
functions directly from the provider but you would probably use them in the Anime
class, but feel free to do it how you want!

Anime
Representing an anime¶

On the last page we learned how to use the provider to search for anime. This page
covers the Anime class, it always represents a single anime.
How to get an Anime object¶

from anipy_api.anime import Anime

provider = ... #

results = provider.get_search("frieren")

anime = []

for r in results:
anime.append(Anime.from_search_result(provider, r))
# or if you really want:
anime.append(Anime(provider, r.name, r.identifier, r.languages))

Get episodes¶

A Episode is either an integer or a float (for .5 episodes).

from anipy_api.provider import LanguageTypeEnum

anime = ...
# List of Episode-type numbers
episodes = anime.get_episodes(lang=LanguageTypeEnum.SUB) #

Get video streams¶

Either use get_video or get_videos, the difference is that one filters out 1 stream
based on quality. Both return ProviderStream objects.

from anipy_api.provider import LanguageTypeEnum

anime = ...

episode_1_stream = anime.get_video(
episode=1,
lang=LanguageTypeEnum.SUB,
preferred_quality=720 #

)
# or get a list of streams that you can filter yourself
episode_1_streams = anime.get_videos(1, LanguageTypeEnum.SUB)

Downloader
Downloading¶

When downloading, you use the Downloader class, it can handle various stream
formats. When downloading you can choose between several download methods:
 download: This method is just a generic method that supports m3u8, mp4 and many
more stream formats. It assumes ffmpeg is installed on the machine please look at
the reference for more info.
m3u8_download: Download a m3u8 playlist.
mp4_download: Download a mp4 stream.
ffmpeg_download: Download any stream supported by ffmpeg. This requires ffmpeg
to be installed on the system!

from pathlib import Path

from anipy_api.download import Downloader

from anipy_api.provider import LanguageTypeEnum

anime = ...
stream = anime.get_video(1, LanguageTypeEnum.SUB, preferred_quality=1080)

def progress_callback(percentage: float): #

print(f"Progress: {percentage:.1f}%", end="\r")

def info_callback(message: str): #

print(f"Message from the downloader: {message}")

def error_callback(message: str): #

s.write(f"Soft error from the downloader: {message}")

downloader = Downloader(progress_callback, info_callback, error_callback)

download_path = downloader.download( #

stream=stream,
download_path=Path("~/Downloads"),
container=".mkv", #

maxRetry=3 #

ffmpeg=False #

MyAnimeList
Why?¶

Uhm, I did not find a proper implementation of the MyAnimeList v2 api for python...
so yeah.
Authentication¶

The MyAnimeList class implement basic authentication via a username/password combo

or via a refresh token.
from anipy_api.mal import MyAnimeList

# password
mal = MyAnimeList.from_password_grant( #

user="test",
password="test",
client_id=None #

)
# or refresh token
mal = MyAnimeList.from_rt_grant(
refresh_token="random-gibberish112mnsd8123109",
client_id="more-random-suff1231283123102938" #

Usage¶

from anipy_api.mal import MyAnimeList, MALMyListStatusEnum

mal = ...

frieren = mal.get_anime(52991)
print(frieren.title)

results = mal.get_search( #

"frieren",
limit=20, #

pages=2 #

)
print(results)

user = mal.get_user() #

print(user)

updated_entry = mal.update_anime_list( #

52991,
status=MALMyListStatusEnum.WATCHING,
watched_episodes=2,
tags=["currently-updated"] #

)
print(updated_entry)

frieren_now = mal.get_anime(52991) #
frieren_now.my_list_status.tags # -> ["currently-updated"]

user_watching = mal.get_anime_list(
status_filter=MALMyListStatusEnum.WATCHING #

)
print(user_list)

mal.remove_from_anime_list(52991)

Adapting between provider anime and MyAnimeList anime¶

Ok so this is the important part. Imagaine you have a provider anime and want to
add that to your myanimelist, but you do not know which mal anime that is. The
MyAnimeListAdapter class can handle that for you.

BUT this does not always work, there is a possibilty that the adapter can not match
the anime, this class uses the Levenshtein Distance algorithm to calculate the
similiarty between names, you can tweak its parameters and also other stuff to
ensure you get a match.

from anipy_api.mal import MyAnimeListAdapter

mal = ...
provider = ... #

adapter = MyAnimeListAdapter(mal, provider)

# Provider -> MyAnimeList

anime = ...
mal_anime = adapter.from_provider(
anime,
minimum_similarity_ratio=0.8, #

use_alternative_names=True #

)
if mal_anime is not None:
print(mal_anime)

# MyAnimeList -> Provider

mal_anime = ...
anime = adapter.from_myanimelist(
mal_anime,
minimum_similarity_ratio=0.8,
use_alternative_names=True,
use_filters=True #

)
if anime is not None:
print(anime)
anilist
AniList(client_id=None) ¶

MyAnimeList api client that implements some of the endpoints documented here.

Attributes:
Name Type Description
API_BASE

The base url of the api (https://api.myanimelist.net/v2)

CLIENT_ID

The client being used to access the api

RESPONSE_FIELDS

Corresponds to fields of AniListAnime object (read here for explaination)

Parameters:
Name Type Description Default
client_id Optional[str]

Overrides the default client id

None

Please note that that currently no complex oauth autentication scheme is

implemented, this client uses the client id of the official MyAnimeList android
app, this gives us the ability to login via a username/password combination. If you
pass your own client id you will not be able to use the from_implicit_grant
function.

Source code in api/src/anipy_api/anilist.py

def __init__(self, client_id: Optional[str] = None):

"""__init__ of AniList.

Args:
client_id: Overrides the default client id

Info:
Please note that that currently no complex oauth autentication scheme is
implemented, this client uses the client id of the official MyAnimeList
android app, this gives us the ability to login via a username/password
combination. If you pass your own client id you will not be able to use
the [from_implicit_grant][anipy_api.anilist.AniList.from_implicit_grant]
function.
"""
if client_id:
self.CLIENT_ID = client_id
self.AUTH_URL = f"https://anilist.co/api/v2/oauth/authorize?
client_id={self.CLIENT_ID}&response_type=token"

self._access_token = None
self._auth_expire_time = datetime.datetime.min
self._session = Session()
self._session.headers.update(
{
"Content-Type": "application/json",
"Accept": "application/json",
 }
)

from_implicit_grant(access_token, client_id=None) staticmethod ¶

Authenticate via Implicit Grant to retrieve access token

Returns:
Type Description
AniList

The AniList client object

Source code in api/src/anipy_api/anilist.py

@staticmethod
def from_implicit_grant(
access_token: str,
client_id: Optional[str] = None
) -> "AniList":
"""Authenticate via Implicit Grant to retrieve access token

Returns:
The AniList client object
"""
anilist = AniList(client_id)
anilist._refresh_auth(access_token)
return anilist

get_anime(anime_id) ¶

Get a MyAnimeList anime by its id.

Parameters:
Name Type Description Default
anime_id int

The id of the anime

required

Returns:
Type Description
AniListAnime

The anime that corresponds to the id

Source code in api/src/anipy_api/anilist.py

def get_anime(self, anime_id: int) -> AniListAnime:

"""Get a MyAnimeList anime by its id.

Args:
anime_id: The id of the anime

Returns:
The anime that corresponds to the id
"""
query = """
query ($id: Int) {
Media (id: $id) {
id
 media_type: format
num_episodes: episodes
title {
user_preferred: userPreferred
}
alternative_titles: title {
english
native
romaji
}
year: seasonYear
season
my_list_status: mediaListEntry {
entry_id: id
notes
num_episodes_watched: progress
status
score
}
}
}
"""
variables = { "id": anime_id }
request = Request("POST", self.API_BASE, json={ 'query': query, 'variables':
variables })
return AniListAnime.from_dict(self._make_request(request)["data"]["Media"])

get_anime_list(status_filter=None) ¶

Get the anime list of the currently authenticated user.

Parameters:
Name Type Description Default
status_filter Optional[AniListMyListStatusEnum]

A filter that determines which list status is retrieved

None

Returns:
Type Description
List[AniListAnime]

List of anime in the anime list

Source code in api/src/anipy_api/anilist.py

def get_anime_list(
self, status_filter: Optional[AniListMyListStatusEnum] = None
) -> List[AniListAnime]:
"""Get the anime list of the currently authenticated user.

Args:
status_filter: A filter that determines which list status is retrieved

Returns:
List of anime in the anime list
"""
query = """
query ($type: MediaType!, $userId: Int!) {
MediaListCollection(type: $type, userId: $userId) {
 lists {
entries {
id
media {
id
media_type: format
num_episodes: episodes
title {
user_preferred: userPreferred
}
alternative_titles: title {
english
native
romaji
}
year: seasonYear
season
my_list_status: mediaListEntry {
entry_id: id
notes
num_episodes_watched: progress
status
score
}
}
}
}
}
}
"""
user_id = self.get_user().id
variables = { "type": "ANIME", "userId": user_id }
request = Request("POST", self.API_BASE, json={ 'query': query, 'variables':
variables })

anime_list = []
for group in self._make_request(request)["data"]["MediaListCollection"]
["lists"]:
for entry in group["entries"]:
anime = AniListAnime.from_dict(entry["media"])
user_status = anime.my_list_status.status if anime.my_list_status else
None
if status_filter is None or user_status == status_filter:
anime_list.append(anime)

return anime_list

get_search(search, limit=20, pages=1) ¶

Search AniList.

Parameters:
Name Type Description Default
search str

Search query
required
limit int
The amount of results per page
20
pages int

The amount of pages to return, note the total number of results is limit times
pages
1

Returns:
Type Description
List[AniListAnime]

A list of search results

Source code in api/src/anipy_api/anilist.py

def get_search(self, search: str, limit: int = 20, pages: int = 1) ->
List[AniListAnime]:
"""Search AniList.

Args:
search: Search query
limit: The amount of results per page
pages: The amount of pages to return,
note the total number of results is limit times pages

Returns:
A list of search results
"""
query = """
query ($search: String!, $page: Int, $perPage: Int) {
Page (page: $page, perPage: $perPage){
page_info: pageInfo {
currentPage
hasNextPage
}
media (search: $search, type: ANIME) {
id
media_type: format
num_episodes: episodes
title {
user_preferred: userPreferred
}
alternative_titles: title {
english
native
romaji
}
year: seasonYear
season
my_list_status: mediaListEntry {
entry_id: id
notes
num_episodes_watched: progress
status
score
}
}
}
}
 """

anime_list = []
next_page = True

for page in range(pages):

if next_page:
variables = { "search": search, "page": page+1, "perPage": limit }
request = Request("POST", self.API_BASE, json={ 'query': query,
'variables': variables })
response = AniListPagingResource.from_dict(self._make_request(request)
["data"]["Page"])
anime_list.extend(response.media)

next_page = response.page_info.has_next_page

return anime_list

get_user() ¶

Get information about the currently authenticated user.

Returns:
Type Description
AniListUser

A object with user information

Source code in api/src/anipy_api/anilist.py

def get_user(self) -> AniListUser:

"""Get information about the currently authenticated user.

Returns:
A object with user information
"""
query="""
query {
Viewer {
id
name
picture: avatar {
large
medium
}
}
}
"""
request = Request(
"POST", self.API_BASE, json={ 'query': query }
)
return AniListUser.from_dict(self._make_request(request)["data"]["Viewer"])

anime
Anime(provider, name, identifier, languages) ¶

A wrapper class that represents an anime, it is pretty useful, but you can also
just use the Provider without the wrapper.

Parameters:
Name Type Description Default
provider BaseProvider

The provider from which the identifier was retrieved

required
name str

The name of the Anime

required
identifier str

The identifier of the Anime

required
languages Set[LanguageTypeEnum]

Supported Language types of the Anime

required

Attributes:
Name Type Description
provider BaseProvider

The from which the Anime comes from

name str

The name of the Anime

identifier str

The identifier of the Anime

languages Set[LanguageTypeEnum]

Set of supported Language types of the Anime

Source code in api/src/anipy_api/anime.py

def __init__(
self,
provider: "BaseProvider",
name: str,
identifier: str,
languages: Set["LanguageTypeEnum"],
):
self.provider: "BaseProvider" = provider
self.name: str = name
self.identifier: str = identifier
self.languages: Set["LanguageTypeEnum"] = languages

from_local_list_entry(entry) staticmethod ¶

Get Anime object from LocalListEntry

Parameters:
Name Type Description Default
entry LocalListEntry

The local list entry

required
Returns:
Type Description
Anime

Anime Object
Source code in api/src/anipy_api/anime.py

@staticmethod
def from_local_list_entry(entry: "LocalListEntry") -> "Anime":
"""Get Anime object from [LocalListEntry][anipy_api.locallist.LocalListEntry]

Args:
entry: The local list entry

Returns:
Anime Object
"""
provider = next(
filter(lambda x: x.NAME == entry.provider, list_providers()), None
)

if provider is None:
raise ProviderNotAvailableError(entry.provider)

return Anime(provider(), entry.name, entry.identifier, entry.languages)

from_search_result(provider, result) staticmethod ¶

Get Anime object from ProviderSearchResult.

Parameters:
Name Type Description Default
provider BaseProvider

The provider from which the search result stems from

required
result ProviderSearchResult

The search result

required

Returns:
Type Description
Anime

Anime object
Source code in api/src/anipy_api/anime.py

@staticmethod
def from_search_result(
provider: "BaseProvider", result: "ProviderSearchResult"
) -> "Anime":
"""Get Anime object from ProviderSearchResult.

Args:
provider: The provider from which the search result stems from
result: The search result

Returns:
 Anime object
"""
return Anime(provider, result.name, result.identifier, result.languages)

get_episodes(lang) ¶

Get a list of episodes from the Anime.

Parameters:
Name Type Description Default
lang LanguageTypeEnum

Language type that determines if episodes are searched for the dub or sub version
of the Anime. Use the languages attribute to get supported languages for this
Anime.
required

Returns:
Type Description
List[Episode]

List of Episodes
Source code in api/src/anipy_api/anime.py

def get_episodes(self, lang: "LanguageTypeEnum") -> List["Episode"]:

"""Get a list of episodes from the Anime.

Args:
lang: Language type that determines if episodes are searched
for the dub or sub version of the Anime. Use the `languages`
attribute to get supported languages for this Anime.

Returns:
List of Episodes
"""
return self.provider.get_episodes(self.identifier, lang)

get_info() ¶

Get information about the Anime.

Returns:
Type Description
ProviderInfoResult

ProviderInfoResult object
Source code in api/src/anipy_api/anime.py

def get_info(self) -> "ProviderInfoResult":

"""Get information about the Anime.

Returns:
ProviderInfoResult object
"""
return self.provider.get_info(self.identifier)

get_video(episode, lang, preferred_quality=None) ¶

Get a video stream for the specified episode, the quality to return is determined
by the preferred_quality argument or if this is not defined by the best quality
found. To get a list of streams use get_videos.

Parameters:
Name Type Description Default
episode Episode

The episode to get the stream for

required
lang LanguageTypeEnum

Language type that determines if streams are searched for the dub or sub version of
the Anime. Use the languages attribute to get supported languages for this Anime.
required
preferred_quality Optional[Union[str, int]]

This may be a integer (e.g. 1080, 720 etc.) or the string "worst" or "best".
None

Returns:
Type Description
ProviderStream

A stream

Source code in api/src/anipy_api/anime.py

def get_video(
self,
episode: Episode,
lang: "LanguageTypeEnum",
preferred_quality: Optional[Union[str, int]] = None,
) -> "ProviderStream":
"""Get a video stream for the specified episode, the quality to return
is determined by the `preferred_quality` argument or if this is not
defined by the best quality found. To get a list of streams use
[get_videos][anipy_api.anime.Anime.get_videos].

Args:
episode: The episode to get the stream for
lang: Language type that determines if streams are searched for
the dub or sub version of the Anime. Use the `languages`
attribute to get supported languages for this Anime.
preferred_quality: This may be a integer (e.g. 1080, 720 etc.)
or the string "worst" or "best".

Returns:
A stream
"""
streams = self.provider.get_video(self.identifier, episode, lang)
streams.sort(key=lambda s: s.resolution + (10 if s.subtitle else 0))

if preferred_quality == "worst":
stream = streams[0]
elif preferred_quality == "best":
stream = streams[-1]
elif preferred_quality is None:
stream = streams[-1]
else:
 stream = next(
filter(lambda s: s.resolution == preferred_quality, streams), None
)

if stream is None:
stream = streams[-1]

return stream

get_videos(episode, lang) ¶

Get a list of video streams for the specified Episode.

Parameters:
Name Type Description Default
episode Episode

The episode to get the streams for

required
lang LanguageTypeEnum

Language type that determines if streams are searched for the dub or sub version of
the Anime. Use the languages attribute to get supported languages for this Anime.
required

Returns:
Type Description
List[ProviderStream]

A list of streams sorted by quality

Source code in api/src/anipy_api/anime.py

def get_videos(
self, episode: Episode, lang: "LanguageTypeEnum"
) -> List["ProviderStream"]:
"""Get a list of video streams for the specified Episode.

Args:
episode: The episode to get the streams for
lang: Language type that determines if streams are searched for
the dub or sub version of the Anime. Use the `languages`
attribute to get supported languages for this Anime.

Returns:
A list of streams sorted by quality
"""
streams = self.provider.get_video(self.identifier, episode, lang)
streams.sort(key=lambda s: s.resolution)

return streams

Downloader(progress_callback=None, info_callback=None, soft_error_callback=None) ¶

Downloader class to download streams retrieved by the Providers.

Parameters:
Name Type Description Default
progress_callback Optional[ProgressCallback]
A callback with an percentage argument, that gets called on download progress.
None
info_callback Optional[InfoCallback]

A callback with an message argument, that gets called on certain events.

None
soft_error_callback Optional[InfoCallback]

A callback with a message argument, when certain events cause a non-fatal error (if
none given, alternative fallback is info_callback).
None
Source code in api/src/anipy_api/download.py

def __init__(
self,
progress_callback: Optional[ProgressCallback] = None,
info_callback: Optional[InfoCallback] = None,
soft_error_callback: Optional[InfoCallback] = None,
):
"""__init__ of Downloader.

Args:
progress_callback: A callback with an percentage argument, that gets called
on download progress.
info_callback: A callback with an message argument, that gets called on
certain events.
soft_error_callback: A callback with a message argument, when certain
events cause a non-fatal error (if none given, alternative fallback is
info_callback).
"""
self._progress_callback: ProgressCallback = progress_callback or (
lambda percentage: None
)
self._info_callback: InfoCallback = info_callback or (
lambda message, exc_info=None: None
)
self._soft_error_callback: InfoCallback = (
soft_error_callback
or info_callback
or (lambda message, exc_info=None: None)
)

self._session = requests.Session()

adapter = HTTPAdapter(max_retries=Retry(connect=3, backoff_factor=0.5))

self._session.mount("http://", adapter)
self._session.mount("https://", adapter)

download(stream, download_path, container=None, ffmpeg=False, max_retry=3,

post_dl_cb=None) ¶

Generic download function that determines the best way to download a specific
stream and downloads it. The suffix should be omitted here, you can instead use the
container argument to remux the stream after the download, note that this will
trigger the progress_callback again. This function assumes that ffmpeg is installed
on the system, because if the stream is neither m3u8 or mp4 it will default to
ffmpeg_download.
Parameters:
Name Type Description Default
stream ProviderStream

The stream to download

required
download_path Path

The path to download the stream to.

required
container Optional[str]

The container to remux the video to if it is not already correctly muxed, the user
must have ffmpeg installed on the system. Containers may include all containers
supported by FFmpeg e.g. ".mp4", ".mkv" etc...
None
ffmpeg bool

Wheter to automatically default to ffmpeg_download for m3u8/hls streams.

False
max_retry int

The amount of times the API can retry the download

3
post_dl_cb Optional[PostDownloadCallback]

Called when completing download, not called when file already exsists
None

Returns:
Type Description
Path

The path of the resulting file

Source code in api/src/anipy_api/download.py

def download(
self,
stream: "ProviderStream",
download_path: Path,
container: Optional[str] = None,
ffmpeg: bool = False,
max_retry: int = 3,
post_dl_cb: Optional[PostDownloadCallback] = None,
) -> Path:
"""Generic download function that determines the best way to download a
specific stream and downloads it. The suffix should be omitted here,
you can instead use the `container` argument to remux the stream after
the download, note that this will trigger the `progress_callback`
again. This function assumes that ffmpeg is installed on the system,
because if the stream is neither m3u8 or mp4 it will default to
[ffmpeg_download][anipy_api.download.Downloader.ffmpeg_download].

Args:
stream: The stream to download
download_path: The path to download the stream to.
container: The container to remux the video to if it is not already
correctly muxed, the user must have ffmpeg installed on the system.
Containers may include all containers supported by FFmpeg e.g. ".mp4",
".mkv" etc...
ffmpeg: Wheter to automatically default to
[ffmpeg_download][anipy_api.download.Downloader.ffmpeg_download] for
m3u8/hls streams.
max_retry: The amount of times the API can retry the download
post_dl_cb: Called when completing download, not called when file already
exsists

Returns:
The path of the resulting file
"""
curr_exc: Exception | None = None
post_dl_cb = post_dl_cb or (lambda path, stream: None)
for i in range(max_retry):
try:
path = self._download_single_try(
stream, download_path, post_dl_cb, container, ffmpeg
)
return path
except DownloadError as e:
self._soft_error_callback(str(e))
curr_exc = e
except Exception as e:
self._soft_error_callback(f"An error occurred during download: {e}")
curr_exc = e
self._soft_error_callback(f"{max_retry-i-1} retries remain")

# Impossible, but to make the type

# checker happy
if curr_exc is None:
raise DownloadError("Unknown error occurred")
# If retrying doesn't work, double it and
# give it to the next exception handler
raise curr_exc

ffmpeg_download(stream, download_path) ¶

Download a stream with FFmpeg, FFmpeg needs to be installed on the system. FFmpeg
will be able to handle about any stream and it is also able to remux the resulting
video. By changing the suffix of the download_path you are able to tell ffmpeg to
remux to a specific container.

Parameters:
Name Type Description Default
stream ProviderStream

The stream
required
download_path Path

The path to download to including a specific suffix.

required

Returns:
Type Description
Path

The download path, this should be the same as the

Path
passed one as ffmpeg will remux to about any container.
Source code in api/src/anipy_api/download.py

def ffmpeg_download(self, stream: "ProviderStream", download_path: Path) -> Path:

"""Download a stream with FFmpeg, FFmpeg needs to be installed on the
system. FFmpeg will be able to handle about any stream and it is also
able to remux the resulting video. By changing the suffix of the
`download_path` you are able to tell ffmpeg to remux to a specific
container.

Args:
stream: The stream
download_path: The path to download to including a specific suffix.

Returns:
The download path, this should be the same as the
passed one as ffmpeg will remux to about any container.
"""

ffprobe = (
FFmpeg(executable="ffprobe")
.option("v", 0)
.option("of", "json")
.option("show_program_version")
)
version = json.loads(ffprobe.execute())
version = [
int(''.join(c for c in v if c.isdigit())) for v in
version["program_version"]["version"].split("-")[0].split(".")
]

if len(version) < 3:
version.append(0)

major_v, minor_v, patch_v = version

extension_picky = major_v >= 7 and minor_v >= 1 and patch_v >= 1

ffprobe = FFmpeg(executable="ffprobe").input(
stream.url, print_format="json", show_format=None
)

if stream.referrer:
ffprobe.option("headers", f"Referer: {stream.referrer}")

if extension_picky:
ffprobe.option("extension_picky", 0)

meta = json.loads(ffprobe.execute())
duration = float(meta["format"]["duration"])

output_options: Dict[str, Any] = {

"c:v": "copy",
"c:a": "copy",
"c:s": "mov_text"
}

if extension_picky:
 output_options.update({"extension_picky": 0})

ffmpeg = (
FFmpeg()
.option("y")
.option("v", "warning")
.option("stats")
.input(stream.url)
.output(
download_path,
output_options
)
)

if stream.referrer:
ffmpeg.option("headers", f"Referer: {stream.referrer}")

@ffmpeg.on("progress")
def on_progress(progress: Progress):
self._progress_callback(progress.time.total_seconds() / duration * 100)

try:
ffmpeg.execute()
except KeyboardInterrupt:
self._info_callback("interrupted deleting partially downloaded file")
download_path.unlink()
raise

return download_path

m3u8_download(stream, download_path) ¶

Download a m3u8/hls stream to a specified download path in a ts container.

The suffix of the download path will be replaced (or added) with ".ts", use the
path returned instead of the passed path.

Parameters:
Name Type Description Default
stream ProviderStream

The m3u8/hls stream

required
download_path Path

The path to save the downloaded stream to

required

Raises:
Type Description
DownloadError

Raised on download error

Returns:
Type Description
Path

The path with a ".ts" suffix

Source code in api/src/anipy_api/download.py

def m3u8_download(self, stream: "ProviderStream", download_path: Path) -> Path:

"""Download a m3u8/hls stream to a specified download path in a ts container.

The suffix of the download path will be replaced (or added) with
".ts", use the path returned instead of the passed path.

Args:
stream: The m3u8/hls stream
download_path: The path to save the downloaded stream to

Raises:
DownloadError: Raised on download error

Returns:
The path with a ".ts" suffix
"""
temp_folder = download_path.parent / "temp"
temp_folder.mkdir(exist_ok=True)
download_path = download_path.with_suffix(".ts")
res = self._session.get(stream.url, headers={"Referer": stream.referrer})
res.raise_for_status()

m3u8_content = m3u8.M3U8(res.text, base_uri=urljoin(res.url, "."))

assert m3u8_content.is_variant is False

counter = 0

def download_ts(segment: m3u8.Segment):

nonlocal counter
url = urljoin(segment.base_uri, segment.uri)
segment_uri = Path(segment.uri)
fname = (
temp_folder / self._get_valid_pathname(segment_uri.stem)
).with_suffix(segment_uri.suffix)
try:
res = self._session.get(str(url), headers={"Referer": stream.referrer})
res.raise_for_status()

with fname.open("wb") as fout:

fout.write(res.content)

counter += 1
self._progress_callback(counter / len(m3u8_content.segments) * 100)
except Exception as e:
# TODO: This gets ignored, because it's in a seperate thread...
raise DownloadError(
f"Encountered this error while downloading: {str(e)}"
)

try:
with ThreadPoolExecutor(max_workers=12) as pool_video:
futures = [
pool_video.submit(download_ts, s) for s in m3u8_content.segments
]
try:
for future in as_completed(futures):
 future.result()
except KeyboardInterrupt:
self._info_callback(
"Download Interrupted, cancelling futures, this may take a
while..."
)
pool_video.shutdown(wait=False, cancel_futures=True)
raise

self._info_callback("Parts Downloaded")

self._info_callback("Merging Parts...")
with download_path.open("wb") as merged:
for segment in m3u8_content.segments:
uri = Path(segment.uri)
fname = (
temp_folder / self._get_valid_pathname(uri.stem)
).with_suffix(uri.suffix)
if not fname.is_file():
raise DownloadError(
f"Could not merge, missing a segment of this playlist:
{stream.url}"
)

with fname.open("rb") as mergefile:

shutil.copyfileobj(mergefile, merged)

self._info_callback("Merge Finished")
shutil.rmtree(temp_folder)

return download_path
except KeyboardInterrupt:
self._info_callback("Download Interrupted, deleting partial file.")
download_path.unlink(missing_ok=True)
shutil.rmtree(temp_folder)
raise