Skip to content

Developer Interface

Funzioni

find

find(keyword)

Ricerca un anime tramite le API interne di Animeworld.

Parameters:

Name Type Description Default
keyword str

Il nome dell'anime o una porzione di esso.

required

Returns:

Type Description
List[Dict]

Informazioni riguardanti l'anime.

Raises:

Type Description
DeprecatedLibrary

Cambiamento del sito Animeworld.

Example
[
  {
    "id": int, # ID interno di AnimeWorld
    "name": str, # Nome dell'anime
    "jtitle": str, # Nome giapponese (con caratteri latini)
    "studio": str, # Studio dell'anime
    "release": str | None, # Giorno, Mese e Anno della release dell'anime. Se non disponibile ("??" al posto della data), ritorna None.
    "episodes": int, # Numero di episodi
    "state": str, # Es. "0", "1", ...
    "story": str, # Trama dell'anime
    "categories": List[dict], # Es. [{"id": int, "name": str, "slug": str, "description": str}]
    "image": str, # Link dell'immagine di copertina
    "durationEpisodes": str, # Durata episodio
    "link": str, # Link dell'anime
    "createdAt": str, # Es. "2021-10-24T18:29:34.000Z"
    "language": str, # Es. "jp
    "year": str, # Anno di rilascio dell'anime
    "dub": bool, # Se è doppiato o meno
    "season": str, # Es. "winter"
    "totViews": int, # Numero totale di visite alla pagina AnimeWolrd
    "dayViews": int, # Numero giornaliero di visite alla pagina AnimeWolrd
    "weekViews": int, # Numero settimanale di visite alla pagina AnimeWolrd
    "monthViews": int, # Numero mensile di visite alla pagina AnimeWolrd
    "malId": int, # ID di MyAnimeList dell'anime
    "anilistId": int, # ID di AniList dell'anime
    "mangaworldId": int, # ID di MangaWorld dell'anime
    "malVote": float, # Valutazione di MyanimeList
    "trailer": str # Link del trailer dell'anime
  }
]
Source code in animeworld/utility.py
@HealthCheck
def find(keyword: str) -> List[Dict]:
    """
    Ricerca un anime tramite le API interne di Animeworld.

    Args:
      keyword: Il nome dell'anime o una porzione di esso.

    Returns:
      Informazioni riguardanti l'anime.

    Raises:
      DeprecatedLibrary: Cambiamento del sito Animeworld.

    Example:
      ```py
      [
        {
          "id": int, # ID interno di AnimeWorld
          "name": str, # Nome dell'anime
          "jtitle": str, # Nome giapponese (con caratteri latini)
          "studio": str, # Studio dell'anime
          "release": str | None, # Giorno, Mese e Anno della release dell'anime. Se non disponibile ("??" al posto della data), ritorna None.
          "episodes": int, # Numero di episodi
          "state": str, # Es. "0", "1", ...
          "story": str, # Trama dell'anime
          "categories": List[dict], # Es. [{"id": int, "name": str, "slug": str, "description": str}]
          "image": str, # Link dell'immagine di copertina
          "durationEpisodes": str, # Durata episodio
          "link": str, # Link dell'anime
          "createdAt": str, # Es. "2021-10-24T18:29:34.000Z"
          "language": str, # Es. "jp
          "year": str, # Anno di rilascio dell'anime
          "dub": bool, # Se è doppiato o meno
          "season": str, # Es. "winter"
          "totViews": int, # Numero totale di visite alla pagina AnimeWolrd
          "dayViews": int, # Numero giornaliero di visite alla pagina AnimeWolrd
          "weekViews": int, # Numero settimanale di visite alla pagina AnimeWolrd
          "monthViews": int, # Numero mensile di visite alla pagina AnimeWolrd
          "malId": int, # ID di MyAnimeList dell'anime
          "anilistId": int, # ID di AniList dell'anime
          "mangaworldId": int, # ID di MangaWorld dell'anime
          "malVote": float, # Valutazione di MyanimeList
          "trailer": str # Link del trailer dell'anime
        }
      ]
      ```
    """

    res = SES.post("https://www.animeworld.so/api/search/v2?", params = {"keyword": keyword}, follow_redirects=True)

    data = res.json()
    if "error" in data: return []
    data = data["animes"]

    for elem in data:
        for k in elem:
            if elem[k] == "??":
                elem[k] = None

    data.sort(key=lambda a: a["dub"])

    return [
        {
        "id": elem["id"],
        "name": elem["name"],
        "jtitle": elem["jtitle"],
        "studio": elem["studio"],
        "release": elem["release"],
        "episodes": int(elem["episodes"]) if elem["episodes"] is not None else None,
        "state": elem["state"],
        "story": elem["story"],
        "categories": elem["categories"],
        "image": elem["image"],
        "durationEpisodes": elem["durationEpisodes"],
        "link": f"https://www.animeworld.so/play/{elem['link']}.{elem['identifier']}" if elem['link'] is not None or elem['identifier'] is not None else None,
        "createdAt": elem["createdAt"],
        "language": elem["language"],
        "year": elem["year"],
        "dub": elem["dub"] != "0" if elem["dub"] is not None else None,
        "season": elem["season"],
        "totViews": elem["totViews"],
        "dayViews": elem["dayViews"],
        "weekViews": elem["weekViews"],
        "monthViews": elem["monthViews"],
        "malId": elem["malId"],
        "anilistId": elem["anilistId"],
        "mangaworldId": elem["mangaworldId"],
        "malVote": elem["malVote"],
        "trailer": elem["trailer"]
        }for elem in data
    ]

Classi

Anime

Anime(link)

Attributes:

Name Type Description
link str

Link dell'anime.

html bytes

Pagina web di Animeworld dell'anime.

Parameters:

Name Type Description Default
link str

Link dell'anime.

required

Raises:

Type Description
DeprecatedLibrary

Cambiamento del sito Animeworld.

Error404

È una pagina 404.

Source code in animeworld/anime.py
def __init__(self, link: str):
    """		
    Args:
      link: Link dell'anime.

    Raises:
      DeprecatedLibrary: Cambiamento del sito Animeworld.
      Error404: È una pagina 404.
    """

    self.link:str = link
    self.html:bytes = self.__getHTML().content
    self.__check404()

getEpisodes

getEpisodes(nums=None)

Ottiene tutti gli episodi dell'anime.

Parameters:

Name Type Description Default
nums Union[List[int], List[str]]

I numeri degli episodi da ottenere

None
Note

Se nums è None o [] allora il metodo restituisce tutti gli episodi dell'anime.

Returns:

Type Description
List[Episodio]

Lista di oggetti Episodio.

Raises:

Type Description
AnimeNotAvailable

L'anime non è ancora disponibile.

DeprecatedLibrary

Cambiamento del sito Animeworld.

Example
return [
  Episodio, # Classe Episodio
  ...
]
Source code in animeworld/anime.py
@HealthCheck
def getEpisodes(self, nums: Union[List[int], List[str]] = None) -> List[Episodio]: # Ritorna una lista di Episodi
    """
    Ottiene tutti gli episodi dell'anime.

    Args:
      nums: I numeri degli episodi da ottenere

    Note:
      Se `nums` è `None` o `[]` allora il metodo restituisce tutti gli episodi dell'anime.

    Returns:
      Lista di oggetti Episodio.

    Raises:
      AnimeNotAvailable: L'anime non è ancora disponibile.
      DeprecatedLibrary: Cambiamento del sito Animeworld.

    Example:
      ```py
      return [
        Episodio, # Classe Episodio
        ...
      ]
      ```
    """

    # Controllo se viene passata una lista di episodi da filtrare
    if nums: nums = list(map(str, nums))

    soupeddata = BeautifulSoup(self.html.decode('utf-8', 'ignore'), "html.parser")

    a_link = soupeddata.select_one('li.episode > a')
    if a_link is None: raise AnimeNotAvailable(self.getName())

    self.link = "https://www.animeworld.so" + a_link.get('href')

    provLegacy = self.__getServer() # vecchio sistema di cattura server

    raw_eps = {}
    for provID in provLegacy:
        prov_soup = soupeddata.select_one(f"div[class*='server'][data-name='{provID}']")

        for data in prov_soup.select('li.episode > a'):
            epNum = data.get('data-episode-num')
            epID = data.get('data-episode-id')

            if epID not in raw_eps:
                raw_eps[epID] = {
                    'number': epNum,
                    'link': f"https://www.animeworld.so/api/download/{epID}",
                    'legacy': [{
                        "id": int(provID),
                        "name": provLegacy[provID]["name"],
                        "link": "https://www.animeworld.so" + data.get("href")
                    }]
                }
            else:
                raw_eps[epID]['legacy'].append({
                "id": int(provID),
                "name": provLegacy[provID]["name"],
                "link": "https://www.animeworld.so" + data.get("href")
            })

    return [
        Episodio(x['number'], x['link'], x['legacy']) 
        for x in list(raw_eps.values())
        if not nums or x['number'] in nums
    ]

getInfo

getInfo()

Ottiene le informazioni dell'anime.

Returns:

Type Description
Dict[str, str]

Informazioni anime.

Raises:

Type Description
DeprecatedLibrary

Cambiamento del sito Animeworld.

Example
return {
  'Categoria': str,
  'Audio': str,
  'Data di Uscita': str,
  'Stagione': str,
  'Studio': str,
  'Genere': List[str],
  'Voto': str,
  'Durata': str,
  'Episodi': str,
  'Stato': str,
  'Visualizzazioni': str
}
Source code in animeworld/anime.py
@HealthCheck
def getInfo(self) -> Dict[str, str]:
    """
    Ottiene le informazioni dell'anime.

    Returns:
      Informazioni anime.

    Raises:
      DeprecatedLibrary: Cambiamento del sito Animeworld.

    Example:
      ```py
      return {
        'Categoria': str,
        'Audio': str,
        'Data di Uscita': str,
        'Stagione': str,
        'Studio': str,
        'Genere': List[str],
        'Voto': str,
        'Durata': str,
        'Episodi': str,
        'Stato': str,
        'Visualizzazioni': str
      }
      ```
    """
    soupeddata = BeautifulSoup(self.html, "html.parser")
    block = soupeddata.find("div", { "class" : "info" }).find("div", { "class" : "row" })

    tName = [x.get_text().replace(':', '') for x in block.find_all("dt")]
    tInfo = []
    for x in block.find_all("dd"):
        txt = x.get_text()
        if len(txt.split(',')) > 1:
            tInfo.append([x.strip() for x in txt.split(',')])
        else:	
            tInfo.append(txt.strip())

    return dict(zip(tName, tInfo))

getName

getName()

Ottiene il nome dell'anime.

Returns:

Type Description
str

Nome anime.

Raises:

Type Description
DeprecatedLibrary

Cambiamento del sito Animeworld.

Example
return str # Nome dell'anime
Source code in animeworld/anime.py
@HealthCheck
def getName(self) -> str: # Nome dell'anime
    """
    Ottiene il nome dell'anime.

    Returns:
      Nome anime.

    Raises:
      DeprecatedLibrary: Cambiamento del sito Animeworld.

    Example:
      ```py
      return str # Nome dell'anime
      ```
    """
    soupeddata = BeautifulSoup(self.html, "html.parser")
    return soupeddata.find("h1", { "id" : "anime-title" }).get_text()

getTrama

getTrama()

Ottiene la trama dell'anime.

Returns:

Type Description
str

La trama dell'anime.

Raises:

Type Description
DeprecatedLibrary

Cambiamento del sito Animeworld.

Example
return str # Trama anime.
Source code in animeworld/anime.py
@HealthCheck
def getTrama(self) -> str:
    """
    Ottiene la trama dell'anime.

    Returns:
      La trama dell'anime.

    Raises:
      DeprecatedLibrary: Cambiamento del sito Animeworld.

    Example:
      ```py
      return str # Trama anime.
      ```
    """
    soupeddata = BeautifulSoup(self.html, "html.parser")
    return soupeddata.find("div", { "class" : "desc" }).get_text()

Episodio

Attributes:

Name Type Description
number str

Numero dell'episodio.

links List[Server]

Lista dei server in cui è hostato l'episodio.

Warning

L'attributo number è di tipo str perchè è possibile che capitino episodi con un numero composto (es. 5.5, 268-269), è un caso molto raro ma possibile.

Source code in animeworld/episodio.py
class Episodio:
    """
    Attributes:
      number: Numero dell'episodio.
      links: Lista dei server in cui è hostato l'episodio.

    Warning:
      L'attributo `number` è di tipo `str` perchè è possibile che capitino episodi con un numero composto (es. `5.5`, `268-269`), è un caso molto raro ma possibile.
    """

    def __init__(self, number: str, link: str, legacy: List[Dict] = []):
        """
        Args:
          number: Numero dell'episodio.
          link: Link dell'endpoint dell'episodio.
          legacy: Lista di tutti i link dei server in cui sono hostati gli episodi.
        """
        self.number:str = number 
        self.__link = link
        self.__legacy = legacy

    @property
    def links(self) -> List[Server]: # lista dei provider dove sono hostati gli ep
        """
        Ottiene la lista dei server in cui è hostato l'episodio.

        Returns:
          Lista di oggetti Server.

        Example:
          ```py
          return [
            Server, # Classe Server
            ...
          ]
          ```
        """
        tmp = [] # tutti i links
        res = SES.post(self.__link, timeout=(3, 27), follow_redirects=True)
        data = res.json()

        for provID in data["links"]:
            key = [x for x in data["links"][provID].keys() if x != 'server'][0]
            tmp.append({
                "id": int(provID),
                "name": data["links"][provID]["server"]["name"],
                "link": data["links"][provID][key]["link"]
            })

        for prov in self.__legacy:
            if str(prov['id']) in data["links"].keys(): continue

            tmp.append(prov)

        return self.__setServer(tmp, self.number)

    def fileInfo(self) -> Dict[str,str]:
        """
        Recupera le informazioni del file dell'episodio.

        Returns:
          Informazioni file episodio.

        Example:
          ```py
          return {
            "content_type": str, # Tipo del file, es. video/mp4
            "total_bytes": int, # Byte totali del file
            "last_modified": datetime, # Data e ora dell'ultimo aggiornamento effettuato all'episodio sul server
            "server_name": str, # Nome del server
            "server_id": int, # ID del server
            "url": str # url dell'episodio
          } 
          ```
        """

        info = ""
        err = None
        for server in self.links:
            try:
                info = server.fileInfo()
            except ServerNotSupported:
                pass
            except httpx.HTTPError as exc:
                err = exc
            else:
                return info

        raise err

    def download(self, title: Optional[str]=None, folder: Union[str, io.IOBase]='', *, hook: Callable[[Dict], None]=lambda *args:None, opt: List[str]=[]) -> Optional[str]: # Scarica l'episodio con il primo link nella lista
        """
        Scarica l'episodio dal server più veloce.

        Args:
          title: Nome con cui verrà nominato il file scaricato.
          folder: Posizione in cui verrà spostato il file scaricato.

        Other parameters:
          hook: Funzione che viene richiamata varie volte durante il download; la funzione riceve come argomento un dizionario con le seguenti chiavi:\n 
            - `total_bytes`: Byte totali da scaricare.
            - `downloaded_bytes`: Byte attualmente scaricati.
            - `percentage`: Percentuale del progresso di download.
            - `speed`: Velocità di download (byte/s)
            - `elapsed`: Tempo trascorso dall'inizio del download.
            - `eta`: Tempo stimato rimanente per fine del download.
            - `status`: 'downloading' | 'finished' | 'aborted'
            - `filename`: Nome del file in download.

          opt: Lista per delle opzioni aggiuntive.\n
            - `'abort'`: Ferma forzatamente il download.

        Returns:
          Nome del file scaricato. 

        Raises:
          HardStoppedDownload: Il file in download è stato forzatamente interrotto.

        Example:
          ```py
          return str # File scaricato
          ```
        """

        return self.__choiceBestServer().download(title,folder,hook=hook,opt=opt)

    # Private
    def __setServer(self, links: List[Dict], numero: str) -> List[Server]: # Per ogni link li posizioni nelle rispettive classi
        """
        Costruisce la rispettiva classe Server per ogni link passato.

        Args:        
          links: Dizionario ('id', 'name', 'link') contenente le informazioni del Server in cui è hostato l'episodio.
          numero: Numero dell'episodio.

        Returns:
          Lista di oggetti Server.

        Example:
          ```py
          return [
            Server, # Classe Server
            ...
          ]
          ```
        """
        ret: List[Server] = [] # lista dei server
        for prov in links:
            if prov["id"] == 4:
                ret.append(YouTube(prov["link"], prov["id"], prov["name"], numero))
            elif prov["id"] == 9:
                ret.append(AnimeWorld_Server(prov["link"], prov["id"], prov["name"], numero))
            elif prov["id"] == 8:
                ret.append(Streamtape(prov["link"], prov["id"], prov["name"], numero))
            else:
                ret.append(Server(prov["link"], prov["id"], prov["name"], numero))
        ret.sort(key=self.__sortServer)
        return ret

    # Private
    def __sortServer(self, elem):
        """
        Ordina i server per importanza.
        """
        if isinstance(elem, YouTube): return 0
        elif isinstance(elem, AnimeWorld_Server): return 1
        elif isinstance(elem, Streamtape): return 2
        else: return 4

    def __choiceBestServer(self) -> Server:
        """
        Sceglie il server più veloce per il download dell'episodio.

        Returns:
          Il Server più veloce.
        """
        servers = self.links

        speed_test = [{
            "server": x,
            "bytes": -1
        } for x in servers]

        max_time = 0.5 # numero di secondi massimo

        for test in speed_test:
            try:
                start = time.perf_counter()
                link = test["server"].fileLink()
                if not link: continue
                with SES.stream("GET", link, timeout=0.9, follow_redirects=True) as r:
                    for chunk in r.iter_bytes(chunk_size = 2048):
                        if time.perf_counter() - start > max_time: break
                        test["bytes"] += len(chunk)
            except (ServerNotSupported, httpx.HTTPError):
                continue

        speed_test = [x for x in speed_test if x["bytes"] != -1] # tolgo tutti i server che hanno generato un eccezione
        if len(speed_test) == 0: return servers[0] # ritorno al caso standard

        return max(speed_test, key=lambda x: x["bytes"])["server"] # restituisco il server che ha scaricato più byte in `max_time` secondi
links

Ottiene la lista dei server in cui è hostato l'episodio.

Returns:

Type Description
List[Server]

Lista di oggetti Server.

Example
return [
  Server, # Classe Server
  ...
]

download

download(title=None, folder='', *, hook=lambda : None, opt=[])

Scarica l'episodio dal server più veloce.

Parameters:

Name Type Description Default
title Optional[str]

Nome con cui verrà nominato il file scaricato.

None
folder Union[str, IOBase]

Posizione in cui verrà spostato il file scaricato.

''

Other Parameters:

Name Type Description
hook Callable[[Dict], None]

Funzione che viene richiamata varie volte durante il download; la funzione riceve come argomento un dizionario con le seguenti chiavi:

  • total_bytes: Byte totali da scaricare.
  • downloaded_bytes: Byte attualmente scaricati.
  • percentage: Percentuale del progresso di download.
  • speed: Velocità di download (byte/s)
  • elapsed: Tempo trascorso dall'inizio del download.
  • eta: Tempo stimato rimanente per fine del download.
  • status: 'downloading' | 'finished' | 'aborted'
  • filename: Nome del file in download.
opt List[str]

Lista per delle opzioni aggiuntive.

  • 'abort': Ferma forzatamente il download.

Returns:

Type Description
Optional[str]

Nome del file scaricato.

Raises:

Type Description
HardStoppedDownload

Il file in download è stato forzatamente interrotto.

Example
return str # File scaricato
Source code in animeworld/episodio.py
def download(self, title: Optional[str]=None, folder: Union[str, io.IOBase]='', *, hook: Callable[[Dict], None]=lambda *args:None, opt: List[str]=[]) -> Optional[str]: # Scarica l'episodio con il primo link nella lista
    """
    Scarica l'episodio dal server più veloce.

    Args:
      title: Nome con cui verrà nominato il file scaricato.
      folder: Posizione in cui verrà spostato il file scaricato.

    Other parameters:
      hook: Funzione che viene richiamata varie volte durante il download; la funzione riceve come argomento un dizionario con le seguenti chiavi:\n 
        - `total_bytes`: Byte totali da scaricare.
        - `downloaded_bytes`: Byte attualmente scaricati.
        - `percentage`: Percentuale del progresso di download.
        - `speed`: Velocità di download (byte/s)
        - `elapsed`: Tempo trascorso dall'inizio del download.
        - `eta`: Tempo stimato rimanente per fine del download.
        - `status`: 'downloading' | 'finished' | 'aborted'
        - `filename`: Nome del file in download.

      opt: Lista per delle opzioni aggiuntive.\n
        - `'abort'`: Ferma forzatamente il download.

    Returns:
      Nome del file scaricato. 

    Raises:
      HardStoppedDownload: Il file in download è stato forzatamente interrotto.

    Example:
      ```py
      return str # File scaricato
      ```
    """

    return self.__choiceBestServer().download(title,folder,hook=hook,opt=opt)

fileInfo

fileInfo()

Recupera le informazioni del file dell'episodio.

Returns:

Type Description
Dict[str, str]

Informazioni file episodio.

Example
return {
  "content_type": str, # Tipo del file, es. video/mp4
  "total_bytes": int, # Byte totali del file
  "last_modified": datetime, # Data e ora dell'ultimo aggiornamento effettuato all'episodio sul server
  "server_name": str, # Nome del server
  "server_id": int, # ID del server
  "url": str # url dell'episodio
} 
Source code in animeworld/episodio.py
def fileInfo(self) -> Dict[str,str]:
    """
    Recupera le informazioni del file dell'episodio.

    Returns:
      Informazioni file episodio.

    Example:
      ```py
      return {
        "content_type": str, # Tipo del file, es. video/mp4
        "total_bytes": int, # Byte totali del file
        "last_modified": datetime, # Data e ora dell'ultimo aggiornamento effettuato all'episodio sul server
        "server_name": str, # Nome del server
        "server_id": int, # ID del server
        "url": str # url dell'episodio
      } 
      ```
    """

    info = ""
    err = None
    for server in self.links:
        try:
            info = server.fileInfo()
        except ServerNotSupported:
            pass
        except httpx.HTTPError as exc:
            err = exc
        else:
            return info

    raise err

Server

Attributes:

Name Type Description
link str

Link del server in cui è hostato l'episodio.

Nid int

ID del server.

name str

Nome del server.

number str

Numero dell'episodio.

Source code in animeworld/servers/Server.py
class Server:
    """
    Attributes:
      link: Link del server in cui è hostato l'episodio.
      Nid: ID del server.
      name: Nome del server.
      number: Numero dell'episodio.
    """

    def __init__(self, link: str, Nid: int, name: str, number: str):
        """
        Args:
          link: Link del server in cui è hostato l'episodio.
          Nid: ID del server.
          name: Nome del server.
          number: Numero dell'episodio.
        """

        self.link:str = link
        self.Nid:int = Nid
        self.name:str = name
        self.number:str = number

        # self._HDR = HDR # Protected 
        self._defTitle = f"{self.number} - {self.name}" # nome del file provvisorio

    def _sanitize(self, title: str) -> str: # Toglie i caratteri illegali per i file
        """
        Rimuove i caratteri illegali per il nome del file.

        Args:
          title: Nome del file.

        Returns:
          Il nome del file sanitizzato.

        Example:
          ```py
          return str # title sanitizzato
          ```
        """
        illegal = ['#','%','&','{','}', '\\','<','>','*','?','/','$','!',"'",'"',':','@','+','`','|','=']
        for x in illegal:
            title = title.replace(x, '')
        return title

    def fileInfo(self) -> dict[str,str]:
        """
        Recupera le informazioni del file dell'episodio.

        Returns:
          Informazioni file episodio.

        Raises:
          ServerNotSupported: Se il server non è supportato.

        Example:
          ```py
          return {
            "content_type": str, # Tipo del file, es. video/mp4
            "total_bytes": int, # Byte totali del file
            "last_modified": datetime, # Data e ora dell'ultimo aggiornamento effettuato all'episodio sul server
            "server_name": str, # Nome del server
            "server_id": int, # ID del server
            "url": str # url dell'episodio
          } 
          ```
        """

        raise ServerNotSupported(self.name)

    def fileLink(self) -> str:
        """
        Recupera il link diretto per il download del file dell'episodio.

        Returns:
            Link diretto.

        Raises:
          ServerNotSupported: Se il server non è supportato.

        Example:
          ```py
          return str # Link del file
          ```
        """

        raise ServerNotSupported(self.name)

    def _fileInfoIn(self) -> Dict[str,str]:
        """
        Recupera le informazioni del file dell'episodio usando httpx.
        """
        url = self.fileLink()

        r = SES.head(url, follow_redirects=True)
        r.raise_for_status()

        return {
            "content_type": r.headers['content-type'],
            "total_bytes": int(r.headers['Content-Length']),
            "last_modified": datetime.strptime(r.headers['Last-Modified'], "%a, %d %b %Y %H:%M:%S %Z"),
            "server_name": self.name,
            "server_id": self.Nid,
            "url": url
        }

    def _fileInfoEx(self) -> Dict[str,str]:
        """
        Recupera le informazioni del file dell'episodio usando yutube_dl.
        """

        class MyLogger(object):
            def debug(self, msg):
                pass
            def warning(self, msg):
                pass
            def error(self, msg):
                print(msg)
                return False

        ydl_opts = {
            'logger': MyLogger()
        }
        with youtube_dl.YoutubeDL(ydl_opts) as ydl:
            url = self.fileLink()
            info = ydl.extract_info(url, download=False)
            return {
                "content_type": "unknown",
                "total_bytes": -1,
                "last_modified": datetime.fromtimestamp(0),
                "server_name": self.name,
                "server_id": self.Nid,
                "url": url
            }


    def download(self, title: Optional[str]=None, folder: Union[str, io.IOBase]='', *, hook: Callable[[Dict], None]=lambda *args:None, opt: List[str]=[]) -> Optional[str]:
        """
        Scarica l'episodio dal primo server funzionante della lista links.

        Args:
          title: Nome con cui verrà nominato il file scaricato.
          folder: Posizione in cui verrà spostato il file scaricato.

        Other parameters:
          hook: Funzione che viene richiamata varie volte durante il download; la funzione riceve come argomento un dizionario con le seguenti chiavi:\n 
            - `total_bytes`: Byte totali da scaricare.
            - `downloaded_bytes`: Byte attualmente scaricati.
            - `percentage`: Percentuale del progresso di download.
            - `speed`: Velocità di download (byte/s)
            - `elapsed`: Tempo trascorso dall'inizio del download.
            - `eta`: Tempo stimato rimanente per fine del download.
            - `status`: 'downloading' | 'finished' | 'aborted'
            - `filename`: Nome del file in download.

          opt: Lista per delle opzioni aggiuntive.\n
            - `'abort'`: Ferma forzatamente il download.

        Returns:
          Nome del file scaricato. 

        Raises:
          ServerNotSupported: Se il server non è supportato.
          HardStoppedDownload: Il file in download è stato forzatamente interrotto.

        Example:
          ```py
          return str # File scaricato
          ```
        """
        raise ServerNotSupported(self.name)

    # Protected
    def _downloadIn(self, title: str, folder: Union[str, io.IOBase], *, hook: Callable[[Dict], None], opt: List[str]) -> Optional[str]: # Scarica l'episodio

        """
        Scarica il file utilizzando httpx.

        Args:
          title: Nome con cui verrà nominato il file scaricato.
          folder: Posizione in cui verrà spostato il file scaricato.

        Other parameters:
          hook: Funzione che viene richiamata varie volte durante il download; la funzione riceve come argomento un dizionario con le seguenti chiavi:\n 
            - `total_bytes`: Byte totali da scaricare.
            - `downloaded_bytes`: Byte attualmente scaricati.
            - `percentage`: Percentuale del progresso di download.
            - `speed`: Velocità di download (byte/s)
            - `elapsed`: Tempo trascorso dall'inizio del download.
            - `eta`: Tempo stimato rimanente per fine del download.
            - `status`: 'downloading' | 'finished' | 'aborted'
            - `filename`: Nome del file in download.

          opt: Lista per delle opzioni aggiuntive.\n
            - `'abort'`: Ferma forzatamente il download.

        Returns:
          Nome del file scaricato. 

        Raises:
          HardStoppedDownload: Il file in download è stato forzatamente interrotto.

        Example:
          ```py
          return str # File scaricato
          ```
        """
        with SES.stream("GET", self.fileLink(), follow_redirects=True) as r:
            r.raise_for_status()
            ext = r.headers['content-type'].split('/')[-1]
            if ext == 'octet-stream': ext = 'mp4'
            file = f"{title}.{ext}"

            total_length = int(r.headers.get('content-length'))
            current_lenght = 0
            start = time.time()
            step = time.time()

            fd:io.IOBase = None
            if isinstance(folder, io.IOBase): fd = folder
            else: fd = open(f"{os.path.join(folder,file)}", 'wb')

            try:
                for chunk in r.iter_bytes(chunk_size = 524288):
                    if chunk: 
                        fd.write(chunk)
                        fd.flush()

                        current_lenght += len(chunk)

                        hook({
                            'total_bytes': total_length,
                            'downloaded_bytes': current_lenght,
                            'percentage': current_lenght/total_length,
                            'speed': len(chunk) / (time.time() - step) if (time.time() - step) != 0 else 0,
                            'elapsed': time.time() - start,
                            'filename': file,
                            'eta': ((total_length - current_lenght) / len(chunk)) * (time.time() - step),
                            'status': 'downloading' if "abort" not in opt else "aborted"
                        })

                        if "abort" in opt: raise HardStoppedDownload(file)

                        step = time.time()

                else:
                    hook({
                        'total_bytes': total_length,
                        'downloaded_bytes': total_length,
                        'percentage': 1,
                        'speed': 0,
                        'elapsed': time.time() - start,
                        'eta': 0,
                        'status': 'finished'
                    })

                    if isinstance(folder, str): fd.close()
                    else: fd.seek(0)
                    return file # Se il file è stato scaricato correttamente
            except HardStoppedDownload:
                if isinstance(folder, str): 
                    fd.close()
                    os.remove(f"{os.path.join(folder,file)}")
                else: fd.seek(0)
                return None

    # Protected
    def _dowloadEx(self, title: str, folder: Union[str, io.IOBase], *, hook: Callable[[Dict], None], opt: List[str]) -> Optional[str]:
        """
        Scarica il file utilizzando yutube_dl.

        Args:
          title: Nome con cui verrà nominato il file scaricato.
          folder: Posizione in cui verrà spostato il file scaricato.

        Other parameters:
          hook: Funzione che viene richiamata varie volte durante il download; la funzione riceve come argomento un dizionario con le seguenti chiavi:\n 
            - `total_bytes`: Byte totali da scaricare.
            - `downloaded_bytes`: Byte attualmente scaricati.
            - `percentage`: Percentuale del progresso di download.
            - `speed`: Velocità di download (byte/s)
            - `elapsed`: Tempo trascorso dall'inizio del download.
            - `eta`: Tempo stimato rimanente per fine del download.
            - `status`: 'downloading' | 'finished' | 'aborted'
            - `filename`: Nome del file in download.

          opt: Lista per delle opzioni aggiuntive.\n
            - `'abort'`: Ferma forzatamente il download.

        Returns:
          Nome del file scaricato. 

        Raises:
          HardStoppedDownload: Il file in download è stato forzatamente interrotto.

        Example:
          ```py
          return str # File scaricato
          ```
        """

        tmp = ''
        if isinstance(folder, str): tmp = folder

        class MyLogger(object):
            def debug(self, msg):
                pass
            def warning(self, msg):
                pass
            def error(self, msg):
                print(msg)
                return False

        def my_hook(d):

            hook({
                'total_bytes': int(d['total_bytes_estimate']),
                'downloaded_bytes': int(d['downloaded_bytes']),
                'percentage': int(d['downloaded_bytes'])/int(d['total_bytes_estimate']),
                'speed': float(d['speed']) if d['speed'] is not None else 0,
                'elapsed': float(d['elapsed']),
                'filename': d['filename'],
                'eta': int(d['eta']),
                'status': d['status'] if "abort" not in opt else "aborted"
            })

            if "abort" in opt: raise HardStoppedDownload(d['filename'])

        ydl_opts = {
            'outtmpl': f"{os.path.join(tmp,title)}.%(ext)s",
            'logger': MyLogger(),
            'progress_hooks': [my_hook],
        }
        with youtube_dl.YoutubeDL(ydl_opts) as ydl:
            url = self.fileLink()
            info = ydl.extract_info(url, download=False)
            filename = ydl.prepare_filename(info)
            try:
                ydl.download([url])
            except HardStoppedDownload:
                os.remove(f"{os.path.join(tmp,filename)}")
                return None
            if isinstance(folder, io.IOBase):
                with open(os.path.join(tmp,filename), 'rb') as f:
                    folder.write(f.read())
                f.seek(0)
                os.remove(f"{os.path.join(tmp,filename)}")
            return filename

download

download(title=None, folder='', *, hook=lambda : None, opt=[])

Scarica l'episodio dal primo server funzionante della lista links.

Parameters:

Name Type Description Default
title Optional[str]

Nome con cui verrà nominato il file scaricato.

None
folder Union[str, IOBase]

Posizione in cui verrà spostato il file scaricato.

''

Other Parameters:

Name Type Description
hook Callable[[Dict], None]

Funzione che viene richiamata varie volte durante il download; la funzione riceve come argomento un dizionario con le seguenti chiavi:

  • total_bytes: Byte totali da scaricare.
  • downloaded_bytes: Byte attualmente scaricati.
  • percentage: Percentuale del progresso di download.
  • speed: Velocità di download (byte/s)
  • elapsed: Tempo trascorso dall'inizio del download.
  • eta: Tempo stimato rimanente per fine del download.
  • status: 'downloading' | 'finished' | 'aborted'
  • filename: Nome del file in download.
opt List[str]

Lista per delle opzioni aggiuntive.

  • 'abort': Ferma forzatamente il download.

Returns:

Type Description
Optional[str]

Nome del file scaricato.

Raises:

Type Description
ServerNotSupported

Se il server non è supportato.

HardStoppedDownload

Il file in download è stato forzatamente interrotto.

Example
return str # File scaricato
Source code in animeworld/servers/Server.py
def download(self, title: Optional[str]=None, folder: Union[str, io.IOBase]='', *, hook: Callable[[Dict], None]=lambda *args:None, opt: List[str]=[]) -> Optional[str]:
    """
    Scarica l'episodio dal primo server funzionante della lista links.

    Args:
      title: Nome con cui verrà nominato il file scaricato.
      folder: Posizione in cui verrà spostato il file scaricato.

    Other parameters:
      hook: Funzione che viene richiamata varie volte durante il download; la funzione riceve come argomento un dizionario con le seguenti chiavi:\n 
        - `total_bytes`: Byte totali da scaricare.
        - `downloaded_bytes`: Byte attualmente scaricati.
        - `percentage`: Percentuale del progresso di download.
        - `speed`: Velocità di download (byte/s)
        - `elapsed`: Tempo trascorso dall'inizio del download.
        - `eta`: Tempo stimato rimanente per fine del download.
        - `status`: 'downloading' | 'finished' | 'aborted'
        - `filename`: Nome del file in download.

      opt: Lista per delle opzioni aggiuntive.\n
        - `'abort'`: Ferma forzatamente il download.

    Returns:
      Nome del file scaricato. 

    Raises:
      ServerNotSupported: Se il server non è supportato.
      HardStoppedDownload: Il file in download è stato forzatamente interrotto.

    Example:
      ```py
      return str # File scaricato
      ```
    """
    raise ServerNotSupported(self.name)

fileInfo

fileInfo()

Recupera le informazioni del file dell'episodio.

Returns:

Type Description
dict[str, str]

Informazioni file episodio.

Raises:

Type Description
ServerNotSupported

Se il server non è supportato.

Example
return {
  "content_type": str, # Tipo del file, es. video/mp4
  "total_bytes": int, # Byte totali del file
  "last_modified": datetime, # Data e ora dell'ultimo aggiornamento effettuato all'episodio sul server
  "server_name": str, # Nome del server
  "server_id": int, # ID del server
  "url": str # url dell'episodio
} 
Source code in animeworld/servers/Server.py
def fileInfo(self) -> dict[str,str]:
    """
    Recupera le informazioni del file dell'episodio.

    Returns:
      Informazioni file episodio.

    Raises:
      ServerNotSupported: Se il server non è supportato.

    Example:
      ```py
      return {
        "content_type": str, # Tipo del file, es. video/mp4
        "total_bytes": int, # Byte totali del file
        "last_modified": datetime, # Data e ora dell'ultimo aggiornamento effettuato all'episodio sul server
        "server_name": str, # Nome del server
        "server_id": int, # ID del server
        "url": str # url dell'episodio
      } 
      ```
    """

    raise ServerNotSupported(self.name)
fileLink()

Recupera il link diretto per il download del file dell'episodio.

Returns:

Type Description
str

Link diretto.

Raises:

Type Description
ServerNotSupported

Se il server non è supportato.

Example
return str # Link del file
Source code in animeworld/servers/Server.py
def fileLink(self) -> str:
    """
    Recupera il link diretto per il download del file dell'episodio.

    Returns:
        Link diretto.

    Raises:
      ServerNotSupported: Se il server non è supportato.

    Example:
      ```py
      return str # Link del file
      ```
    """

    raise ServerNotSupported(self.name)

Last update: June 29, 2023