Package `niconico_dl`

PyPI PyPI - Downloads

niconico_dl

ニコニコ動画にある動画をダウンロードするためのPython用のライブラリです。

警告！
このniconico_dlは開発が停止して、アップデートもこれからありません。
代わりにこちらを使用してください。

リファレンス：https://tasuren.github.io/niconico-dl/
GitHub：https://github.com/tasuren/niconico-dl/

Install

pip install niconico_dl

Examples

Normal

url = "https://www.nicovideo.jp/watch/sm38533566"

with niconico_dl.NicoNicoVideo(url, log=True) as nico:
    data = nico.get_info()
    nico.download(data["video"]["title"] + ".mp4")

print("Downloaded!")

Async

async def start_async():
    url = "https://www.nicovideo.jp/watch/sm9664372"
    async with niconico_dl.NicoNicoVideoAsync(url, log=True) as nico:
        data = await nico.get_info()
        await nico.download(data["video"]["title"] + ".mp4")
    print("Downloaded!")


asyncio.run(start_async())

Command Line

使用方法：niconico_dl [URL]
ダウンロードした動画はoutput.mp4という名前で実行したディレクトリに保存されます。

Notes

もしDiscordのボイスチャットにニコニコ動画を流したい人はNicoNicoVideoAsync.download()ではなくNicoNicoVideoAsync.get_download_link()を使用して取得したダウンロードリンクで流すことを推奨します。
downloadは動画をダウンロードするため時間がかかります。なのでget_download_linkでダウンロードリンクを取得してそれを使い直接流すのを推奨します。
注意：closeをお忘れなく、詳細はリファレンスを見てください。

Expand source code

"""
.. include:: ../README.md
"""

from .async_video_manager import *
from .video_manager import *


__all__ = ("HEADERS", "NicoNicoAcquisitionFailed",
           "NicoNicoVideoAsync", "NicoNicoVideo")
__author__ = "tasuren"
__version__ = "2.2.8"

Sub-modules

niconico_dl.async_video_manager
niconico_dl.templates
niconico_dl.video_manager

Classes

class NicoNicoAcquisitionFailed (*args, **kwargs)

ニコニコ動画から情報を取得するのに失敗した際に発生するもの。

Expand source code

class NicoNicoAcquisitionFailed(Exception):
    """ニコニコ動画から情報を取得するのに失敗した際に発生するもの。"""
    pass

Ancestors

builtins.Exception
builtins.BaseException

class NicoNicoVideo (url: str, log: bool = False, headers: Optional[dict] = None)

ニコニコ動画の情報や動画を取得するためのクラスです。
このクラスから動画データの取得やダウンロードが行なえます。

Parameters

url : str: ニコニコ動画のURLです。
log : bool, default False: ログ出力をするかどうかです。
headers : dict, Optional: 通信時に使用するヘッダーです。
指定されない場合はniconico_dl.HEADERSを使用するので普通は変えなくて大丈夫です。

Attributes

heartbeat_thread : threading.Thread: Heartbeatのスレッドです。
Heartbeatを動かすまでこれはNoneです。

class NicoNicoVideo:
    """ニコニコ動画の情報や動画を取得するためのクラスです。  
    このクラスから動画データの取得やダウンロードが行なえます。

    Parameters
    ----------
    url : str
        ニコニコ動画のURLです。
    log : bool, default False
        ログ出力をするかどうかです。
    headers : dict, Optional
        通信時に使用するヘッダーです。  
        指定されない場合は`niconico_dl.HEADERS`を使用するので普通は変えなくて大丈夫です。

    Attributes
    ----------
    heartbeat_thread : threading.Thread
        Heartbeatのスレッドです。  
        Heartbeatを動かすまでこれはNoneです。

    SeeAlso
    -------
    NicoNicoVideoAsync : このクラスの非同期バージョンです。"""
    def __init__(
        self, url: str, log: bool = False, headers: Optional[dict] = None
    ):
        self._headers = headers or HEADERS
        self.heartbeat_thread: Thread = None

        if "nico.ms" in url:
            url = url.replace("nico.ms/", "www.nicovideo.jp/watch/")
        
        if "sp" in url:
            url = url.replace("sp", "www")

        self._url, self._log = url, log
        self._data, self._download_link = {}, None
        self._working_heartbeat = False
        self._stop = False

    def print(self, *args, first: str = "", **kwargs) -> None:
        """niconico_dl用に用意したログ出力用の`print`です。"""
        if self._log:
            print(
                f"{first}[niconico_dl]",
                *args, **kwargs
            )

    def __enter__(self):
        # `with`構文の最初に呼び出されるもの。
        # Heartbnneatを動かします。
        self.connect()
        return self

    def __exit__(self, exc_type, exc, tb):
        # `with`構文から抜けた際に呼び出されるもの。
        # Heartbeatをストップします。
        self.close()

    def connect(self) -> None:
        """ニコニコ動画に接続します。  
        `download`を使用するにはこれを実行してからする必要があります。  
        そしてニコニコ動画との通信が終了したのなら`close`を実行する必要があります。

        Notes
        -----
        これと`close`は以下のように`with`構文で省略が可能です。  
        ```python
        url = "https://www.nicovideo.jp/watch/sm15713037"
        with niconico_dl.NicoNicoVideo(url) as nico:
            nico.download("video.mp4")
        ```"""
        self.heartbeat_thread = Thread(
            target=self._heartbeat,
            name="niconico_dl.heartbeat"
        )
        self.heartbeat_thread.start()
        self.wait_until_working_heartbeat()

    def close(self, close_loop: bool = True) -> None:
        """NicoNicoVideoAsyncを終了します。  
        動画のダウンロードに必要な通信をするHeartbeatを止めます。  
        もし動画のダウンロードが終わったのならこれを実行してください。  
        `with`構文を使用するのならこれを実行する必要はありません。  
        `with`構文の使用例は`connect`の説明にあります。"""
        self._stop = True
        self.heartbeat_thread.join()

    def get_info(self) -> dict:
        """ニコニコ動画のウェブページから動画のデータを取得する関数です。

        Returns
        -------
        data : dict
            取得した動画の情報です。

        Raises
        ------
        NicoNicoAcquisitionFailed
            ニコニコ動画から情報を取得するのに失敗した際に発生します。"""
        self.print("Getting video data...")
        if not self._data:
            # もし動画データを取得していないなら動画URLのHTMLから動画データを取得する。
            # Heartbeatの通信にも必要なものでもあります。
            soup = BeautifulSoup(
                requests.get(self._url, headers=self._headers[2]).text,
                "html.parser"
            )
            data = soup.find(
                "div", {"id": "js-initial-watch-data"}
            ).get("data-api-data")
            if data:
                self._data = loads(data)
            else:
                raise NicoNicoAcquisitionFailed("ニコニコ動画から情報を取得するのに失敗しました。")
        self.print("Done.")
        return self._data

    def wait_until_working_heartbeat(self) -> None:
        """Heartbeatが動き出すまで待機します。"""
        while not self._working_heartbeat:
            pass

    def is_working_heartbeat(self) -> bool:
        """Heartbeatが動いているかの真偽値を返します。

        Returns
        -------
        is_working : bool
            Heartbeatが動いているかどうかの真偽値です。"""
        return self._working_heartbeat

    def get_download_link(self) -> str:
        """
        ニコニコ動画の動画のダウンロードリンクを取得します。  
        返されるリンクからはmp4の動画をダウンロードできます。

        Returns
        -------
        str : Download link

        Warnings
        --------
        返されるダウンロードリンクはHeartbeatが動いている間でしか使うことができません。  
        ですのでHeartbeatを止める`close`はダウンロードリンクの使用が終わってから実行しましょう。  
        したがって`with`構文を使用して取得したダウンロードリンクはすぐに使えなくなることがあるので注意してください。
        """
        if self._download_link is None:
            # Heartbeatが動いていないなら動かす。
            if not self.is_working_heartbeat():
                self.connect()
            # Heartbeatが動画のURLを取得するまで待機する。
            self.wait_until_working_heartbeat()
            self._download_link = self.result_data["content_uri"]

        return self._download_link

    def download(self, path: str, load_chunk_size: int = 1024) -> None:
        """ニコニコ動画の動画をダウンロードします。  
        mp4形式でダウンロードされます。

        Examples
        --------
        ```python
        url = "https://www.nicovideo.jp/watch/sm1097445"
        with NicoNicoVideo(url) as nico:
            data = nico.get_info()
            title = data["video"]["title"]
            nico.download(title + ".mp4")
        ```

        Notes
        -----
        もし`with`構文を使用しないでダウンロードする場合は、ダウンロード前に`connect`を、ダウンロード終了後に`close`を実行してください。  
        動画データの通信に必要なHeartbeatが永遠に動き続けることになります。

        Parameters
        ----------
        path : str
            ダウンロードするニコニコ動画の動画の保存先です。
        load_chunk_size : int, default 1024
            一度にどれほどの量をダウンロードするかです。"""
        self.print("Now loading...")
        url = self.get_download_link()

        params = (
            (
                "ht2_nicovideo",
                self.result_data["content_auth"]["content_auth_info"]["value"]
            ),
        )
        headers = self._headers[1]
        headers["Content-Type"] = "video/mp4"

        BASE = "Downloading video... :"
        self.print(BASE, "Now loading...", first="\r", end="")

        size = int(
            requests.head(
                url, headers=headers, params=params
            ).headers.get("content-length")
        )

        r = requests.get(url, headers=headers, params=params, stream=True)
        r.raise_for_status()
        now_size = 0

        with open(path, "wb") as f:
            for chunk in r.iter_content(chunk_size=load_chunk_size):
                if chunk:
                    now_size += len(chunk)
                    f.write(chunk)
                    self.print(
                        BASE,
                        f"{int(now_size/size*100)}% ({now_size}/{size})",
                        first="\r", end=""
                    )

        self.print("Done.", first="\n")

    def _heartbeat(self, mode = "http_output_download_parameters") -> None:
        # Heartbeatです。
        self.print("Starting heartbeat...")
        if not self._data:
            self.get_info()

        # セッションに必要なデータを`NicoNicoVideoAsync.get_info`で取得したデータから取得します。
        data = _make_sessiondata(
            self._data["media"]["delivery"]["movie"], mode=mode
        )
        self.print("Sending Heartbeat Init Data... :", data)

        # 一番最初のHeartbeatの通信をします。
        r = requests.post(
            URLS["base_heartbeat"] + "?_format=json",
            headers=self._headers[1], data=dumps(data)
        )
        self.result_data = r.json()["data"]["session"]
        session_id = self.result_data["id"]

        self.print("Done. session_id. : " + str(session_id))
        self._working_heartbeat = True

        data, first = {"session": self.result_data}, True
        get_interval = lambda now: now + data["session"]["keep_method"]["heartbeat"]["lifetime"] / 1000 - 3
        make_url = lambda session_id: f"{URLS['base_heartbeat']}/{session_id}?_format=json&_method=PUT"
        after = get_interval(time())

        while not self._stop:
            now = time()
            # ここで定期的にHeartbeatを送ります。
            if now >= after:
                self.print("Sending heartbeat...", data)

                if first:
                    # 最初は普通とは違うものを先にリクエストする必要があるのでそれをリクエストする。
                    r = requests.options(make_url(session_id), headers=self._headers[0])
                    r.raise_for_status()
                    first = False

                r = requests.post(
                    make_url(session_id), headers=self._headers[1],
                    data=dumps(data)
                )
                self.result_data = r.json()["data"]["session"]

                self.print("Done.")
                data = {"session": self.result_data}
                self.print("Received data", data)

                after = get_interval(now)
            else:
                sleep(0.05)

Methods

def close(self, close_loop: bool = True) ‑> None

NicoNicoVideoAsyncを終了します。
動画のダウンロードに必要な通信をするHeartbeatを止めます。
もし動画のダウンロードが終わったのならこれを実行してください。
with構文を使用するのならこれを実行する必要はありません。
with構文の使用例はconnectの説明にあります。

Expand source code

def close(self, close_loop: bool = True) -> None:
    """NicoNicoVideoAsyncを終了します。  
    動画のダウンロードに必要な通信をするHeartbeatを止めます。  
    もし動画のダウンロードが終わったのならこれを実行してください。  
    `with`構文を使用するのならこれを実行する必要はありません。  
    `with`構文の使用例は`connect`の説明にあります。"""
    self._stop = True
    self.heartbeat_thread.join()

def connect(self) ‑> None

ニコニコ動画に接続します。
downloadを使用するにはこれを実行してからする必要があります。
そしてニコニコ動画との通信が終了したのならcloseを実行する必要があります。

Notes

これとcloseは以下のようにwith構文で省略が可能です。

url = "https://www.nicovideo.jp/watch/sm15713037"
with niconico_dl.NicoNicoVideo(url) as nico:
    nico.download("video.mp4")

Expand source code

def connect(self) -> None:
    """ニコニコ動画に接続します。  
    `download`を使用するにはこれを実行してからする必要があります。  
    そしてニコニコ動画との通信が終了したのなら`close`を実行する必要があります。

    Notes
    -----
    これと`close`は以下のように`with`構文で省略が可能です。  
    ```python
    url = "https://www.nicovideo.jp/watch/sm15713037"
    with niconico_dl.NicoNicoVideo(url) as nico:
        nico.download("video.mp4")
    ```"""
    self.heartbeat_thread = Thread(
        target=self._heartbeat,
        name="niconico_dl.heartbeat"
    )
    self.heartbeat_thread.start()
    self.wait_until_working_heartbeat()

def download(self, path: str, load_chunk_size: int = 1024) ‑> None

ニコニコ動画の動画をダウンロードします。
mp4形式でダウンロードされます。

Examples

url = "https://www.nicovideo.jp/watch/sm1097445"
with NicoNicoVideo(url) as nico:
    data = nico.get_info()
    title = data["video"]["title"]
    nico.download(title + ".mp4")

Notes

もしwith構文を使用しないでダウンロードする場合は、ダウンロード前にconnectを、ダウンロード終了後にcloseを実行してください。
動画データの通信に必要なHeartbeatが永遠に動き続けることになります。

Parameters

path : str: ダウンロードするニコニコ動画の動画の保存先です。
load_chunk_size : int, default 1024: 一度にどれほどの量をダウンロードするかです。

Expand source code

def download(self, path: str, load_chunk_size: int = 1024) -> None:
    """ニコニコ動画の動画をダウンロードします。  
    mp4形式でダウンロードされます。

    Examples
    --------
    ```python
    url = "https://www.nicovideo.jp/watch/sm1097445"
    with NicoNicoVideo(url) as nico:
        data = nico.get_info()
        title = data["video"]["title"]
        nico.download(title + ".mp4")
    ```

    Notes
    -----
    もし`with`構文を使用しないでダウンロードする場合は、ダウンロード前に`connect`を、ダウンロード終了後に`close`を実行してください。  
    動画データの通信に必要なHeartbeatが永遠に動き続けることになります。

    Parameters
    ----------
    path : str
        ダウンロードするニコニコ動画の動画の保存先です。
    load_chunk_size : int, default 1024
        一度にどれほどの量をダウンロードするかです。"""
    self.print("Now loading...")
    url = self.get_download_link()

    params = (
        (
            "ht2_nicovideo",
            self.result_data["content_auth"]["content_auth_info"]["value"]
        ),
    )
    headers = self._headers[1]
    headers["Content-Type"] = "video/mp4"

    BASE = "Downloading video... :"
    self.print(BASE, "Now loading...", first="\r", end="")

    size = int(
        requests.head(
            url, headers=headers, params=params
        ).headers.get("content-length")
    )

    r = requests.get(url, headers=headers, params=params, stream=True)
    r.raise_for_status()
    now_size = 0

    with open(path, "wb") as f:
        for chunk in r.iter_content(chunk_size=load_chunk_size):
            if chunk:
                now_size += len(chunk)
                f.write(chunk)
                self.print(
                    BASE,
                    f"{int(now_size/size*100)}% ({now_size}/{size})",
                    first="\r", end=""
                )

    self.print("Done.", first="\n")

def get_download_link(self) ‑> str

ニコニコ動画の動画のダウンロードリンクを取得します。
返されるリンクからはmp4の動画をダウンロードできます。

Returns

str : Download link

Warnings

返されるダウンロードリンクはHeartbeatが動いている間でしか使うことができません。
ですのでHeartbeatを止めるcloseはダウンロードリンクの使用が終わってから実行しましょう。
したがってwith構文を使用して取得したダウンロードリンクはすぐに使えなくなることがあるので注意してください。

Expand source code

def get_download_link(self) -> str:
    """
    ニコニコ動画の動画のダウンロードリンクを取得します。  
    返されるリンクからはmp4の動画をダウンロードできます。

    Returns
    -------
    str : Download link

    Warnings
    --------
    返されるダウンロードリンクはHeartbeatが動いている間でしか使うことができません。  
    ですのでHeartbeatを止める`close`はダウンロードリンクの使用が終わってから実行しましょう。  
    したがって`with`構文を使用して取得したダウンロードリンクはすぐに使えなくなることがあるので注意してください。
    """
    if self._download_link is None:
        # Heartbeatが動いていないなら動かす。
        if not self.is_working_heartbeat():
            self.connect()
        # Heartbeatが動画のURLを取得するまで待機する。
        self.wait_until_working_heartbeat()
        self._download_link = self.result_data["content_uri"]

    return self._download_link

def get_info(self) ‑> dict

ニコニコ動画のウェブページから動画のデータを取得する関数です。

Returns

data : dict: 取得した動画の情報です。

Raises

NicoNicoAcquisitionFailed: ニコニコ動画から情報を取得するのに失敗した際に発生します。

Expand source code

def get_info(self) -> dict:
    """ニコニコ動画のウェブページから動画のデータを取得する関数です。

    Returns
    -------
    data : dict
        取得した動画の情報です。

    Raises
    ------
    NicoNicoAcquisitionFailed
        ニコニコ動画から情報を取得するのに失敗した際に発生します。"""
    self.print("Getting video data...")
    if not self._data:
        # もし動画データを取得していないなら動画URLのHTMLから動画データを取得する。
        # Heartbeatの通信にも必要なものでもあります。
        soup = BeautifulSoup(
            requests.get(self._url, headers=self._headers[2]).text,
            "html.parser"
        )
        data = soup.find(
            "div", {"id": "js-initial-watch-data"}
        ).get("data-api-data")
        if data:
            self._data = loads(data)
        else:
            raise NicoNicoAcquisitionFailed("ニコニコ動画から情報を取得するのに失敗しました。")
    self.print("Done.")
    return self._data

def is_working_heartbeat(self) ‑> bool

Heartbeatが動いているかの真偽値を返します。

Returns

is_working : bool: Heartbeatが動いているかどうかの真偽値です。

Expand source code

def is_working_heartbeat(self) -> bool:
    """Heartbeatが動いているかの真偽値を返します。

    Returns
    -------
    is_working : bool
        Heartbeatが動いているかどうかの真偽値です。"""
    return self._working_heartbeat

def print(self, *args, first: str = '', **kwargs) ‑> None

niconico_dl用に用意したログ出力用のprintです。

Expand source code

def print(self, *args, first: str = "", **kwargs) -> None:
    """niconico_dl用に用意したログ出力用の`print`です。"""
    if self._log:
        print(
            f"{first}[niconico_dl]",
            *args, **kwargs
        )

def wait_until_working_heartbeat(self) ‑> None

Heartbeatが動き出すまで待機します。

Expand source code

def wait_until_working_heartbeat(self) -> None:
    """Heartbeatが動き出すまで待機します。"""
    while not self._working_heartbeat:
        pass

class NicoNicoVideoAsync (url: str, log: bool = False, headers: Optional[dict] = None, loop: Optional[asyncio.events.AbstractEventLoop] = None)

ニコニコ動画の情報や動画を取得するためのクラスです。
このクラスから動画データの取得やダウンロードが行なえます。
ですがこれは非同期版です。(通常はNicoNicoVideoです。)

Parameters

url : str: ニコニコ動画のURLです。
log : bool, default False: ログ出力をするかどうかです。
headers : dict, optional: 通信時に使用するヘッダーです。
デフォルトはniconico_dl.HEADERSが使われるので普通は変えなくて大丈夫です。
loop : asyncio.AbstractLoop, optional: 使用するイベントループです。
指定しない場合はasyncio.get_event_loopによって自動で取得されます。

Attributes

loop : asyncio.AbstractEventLoop: 使用しているイベントループです。
heartbeat_task : asyncio.Task: HeartbeatのTaskです。
Heartbeatを動かすまでこれはNoneです。

class NicoNicoVideoAsync:
    """ニコニコ動画の情報や動画を取得するためのクラスです。  
    このクラスから動画データの取得やダウンロードが行なえます。  
    ですがこれは非同期版です。(通常は`NicoNicoVideo`です。)

    Parameters
    ----------
    url : str
        ニコニコ動画のURLです。
    log : bool, default False
        ログ出力をするかどうかです。
    headers : dict, optional
        通信時に使用するヘッダーです。  
        デフォルトは`niconico_dl.HEADERS`が使われるので普通は変えなくて大丈夫です。
    loop : asyncio.AbstractLoop, optional
        使用するイベントループです。  
        指定しない場合は`asyncio.get_event_loop`によって自動で取得されます。

    Attributes
    ----------
    loop : asyncio.AbstractEventLoop
        使用しているイベントループです。
    heartbeat_task : asyncio.Task
        HeartbeatのTaskです。  
        Heartbeatを動かすまでこれはNoneです。

    SeeAlso
    -------
    NicoNicoVideo : このクラスの非非同期版です。"""
    def __init__(
        self, url: str, log: bool = False, headers: Optional[dict] = None,
        loop: Optional[asyncio.AbstractEventLoop] = None
    ):
        self.loop: asyncio.AbstractEventLoop = loop or asyncio.get_event_loop()
        self._headers = headers or HEADERS
        self._download_link = None
        self.heartbeat_task: asyncio.Task = None

        if "nico.ms" in url:
            url = url.replace("nico.ms/", "www.nicovideo.jp/watch/")
        
        if "sp" in url:
            url = url.replace("sp", "www")

        self._url, self._log = url, log
        self._data, self._download_link = {}, None
        self._working_heartbeat = asyncio.Event()
        self._stop = False

    def print(self, *args, first: str = "", **kwargs) -> None:
        """niconico_dl用に用意したログ出力用の`print`です。"""
        if self._log:
            print(
                f"{first}[niconico_dl_async]",
                *args, **kwargs
            )

    async def __aenter__(self):
        # `async with`構文の最初に呼び出されるもの。
        # Heartbnneatを動かします。
        await self.connect()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        # `async with`構文から抜けた際に呼び出されるもの。
        # Heartbeatをストップします。
        self.close()

    async def connect(self) -> None:
        """ニコニコ動画に接続します。  
        `download`を使用するにはこれを実行してからする必要があります。  
        そしてニコニコ動画との通信が終了したのなら`close`を実行する必要があります。

        Notes
        -----
        これと`close`は以下のように`async with`構文で省略が可能です。  
        ```python
        url = "https://www.nicovideo.jp/watch/sm9664372"
        async with niconico_dl.NicoNicoVideoAsync(url) as nico:
            await nico.download("video.mp4")
        ```"""
        self.heartbeat_task = self.loop.create_task(
            self._heartbeat(), name="niconico_dl.heartbeat"
        )
        await self.wait_until_working_heartbeat()

    def close(self) -> None:
        """NicoNicoVideoAsyncを終了します。  
        動画のダウンロードに必要な通信をするHeartbeatを止めます。  
        もし動画のダウンロードが終わったのならこれを実行してください。  
        `async with`構文を使用するのならこれを実行する必要はありません。  
        `async with`構文の使用例は`connect`にあります。

        Warnings
        --------
        これを使用してもイベントループは閉じません。"""
        self._stop = True
        if self.heartbeat_task:
            try:
                self.heartbeat_task.cancel()
            except Exception as e:
                if not isinstance(e, asyncio.CancelledError):
                    raise e

    async def get_info(self) -> dict:
        """ニコニコ動画のウェブページから動画のデータを取得するコルーチン関数です。

        Returns
        -------
        data : dict
            取得した動画の情報です。

        Raises
        ------
        NicoNicoAcquisitionFailed
            ニコニコ動画から情報を取得するのに失敗した際に発生します。"""
        self.print("Getting video data...")
        if not self._data:
            # もし動画データを取得していないなら動画URLのHTMLから動画データを取得する。
            # Heartbeatの通信にも必要なものでもあります。
            async with ClientSession(raise_for_status=True) as session:
                async with session.get(self._url, headers=self._headers[2]) as r:
                    soup = BeautifulSoup(await r.text(), "html.parser")
                    data = soup.find(
                        "div", {"id": "js-initial-watch-data"}
                    ).get("data-api-data")
                    if data:
                        self._data = loads(data)
                    else:
                        raise NicoNicoAcquisitionFailed(
                            "ニコニコ動画から情報を取得するのに失敗しました。"
                        )
        self.print("Done.")
        return self._data

    async def wait_until_working_heartbeat(self) -> None:
        """Heartbeatが動き出すまで待機します。"""
        await self._working_heartbeat.wait()

    def is_working_heartbeat(self) -> bool:
        """Heartbeatが動いているかの真偽値を返します。

        Returns
        -------
        is_working : bool
            Heartbeatが動いているかどうかの真偽値です。"""
        return self._working_heartbeat.is_set()

    async def get_download_link(self) -> str:
        """
        ニコニコ動画の動画のダウンロードリンクを取得します。  
        返されるリンクからはmp4の動画をダウンロードできます。

        Returns
        -------
        str : Download link

        Warnings
        --------
        返されるダウンロードリンクはHeartbeatが動いている間でしか使うことができません。  
        ですのでHeartbeatを止める`close`はダウンロードリンクの使用が終わってから実行しましょう。  
        したがって`async with`構文を使用して取得したダウンロードリンクはすぐに使えなくなることがあるので注意してください。
        """
        if self._download_link is None:
            # Heartbeatが動いていないなら動かす。
            if not self.is_working_heartbeat():
                await self.connect()
            # Heartbeatが動画のURLを取得するまで待機する。
            await self.wait_until_working_heartbeat()
            self._download_link = self.result_data["content_uri"]

        return self._download_link

    async def download(self, path: str, load_chunk_size: int = 1024) -> None:
        """ニコニコ動画の動画をダウンロードします。  
        mp4形式でダウンロードされます。

        Examples
        --------
        ```python
        url = "https://www.nicovideo.jp/watch/sm9720246"
        async with NicoNicoVideoAsync(url) as nico:
            data = await nico.get_info()
            title = data["video"]["title"]
            await nico.download(title + ".mp4")
        ```

        Notes
        -----
        もし`async with`構文を使用しないでダウンロードする場合は、ダウンロード前に`connect`を、ダウンロード終了後に`close`を実行してください。  
        動画データの通信に必要なHeartbeatが永遠に動き続けることになります。

        Parameters
        ----------
        path : str
            ダウンロードするニコニコ動画の動画の保存先です。
        load_chunk_size : int, default 1024
            一度にどれほどの量をダウンロードするかです。"""
        self.print("Now loading...")
        url = await self.get_download_link()

        params = (
            (
                "ht2_nicovideo",
                self.result_data["content_auth"]["content_auth_info"]["value"]
            ),
        )
        headers = self._headers[1]
        headers["Content-Type"] = "video/mp4"

        BASE = "Downloading video... :"
        self.print(BASE, "Now loading...", first="\r", end="")

        async with ClientSession(raise_for_status=True) as session:
            async with session.get(url, headers=headers, params=params) as r:
                size, now_size = r.content_length, 0

                self.print(BASE, "Making a null file...", first="\r", end="")

                async with async_open(path, "wb") as f:
                    await f.write(b"")
                async with async_open(path, "ab") as f:
                    async for chunk in r.content.iter_chunked(load_chunk_size):
                        if chunk:
                            now_size += len(chunk)
                            await f.write(chunk)
                            self.print(
                                BASE,
                                f"{int(now_size/size*100)}% ({now_size}/{size})",
                                first="\r", end=""
                            )

        self.print("Done.")

    async def _heartbeat(self, mode = "http_output_download_parameters") -> None:
        # Heartbeatです。
        self.print("Starting heartbeat...")
        if not self._data:
            await self.get_info()

        # セッションに必要なデータを`NicoNicoVideoAsync.get_info`で取得したデータから取得します。
        data = _make_sessiondata(
            self._data["media"]["delivery"]["movie"], mode=mode
        )
        self.print("Sending Heartbeat Init Data... :", data)

        # 一番最初のHeartbeatの通信をします。
        async with ClientSession(raise_for_status=True) as session:
            async with session.post(
                URLS["base_heartbeat"] + "?_format=json",
                headers=self._headers[1], json=data
            ) as r:
                self.result_data = (await r.json(loads=loads))["data"]["session"]
        session_id = self.result_data["id"]

        self.print("Done. session_id. : " + str(session_id))
        self._working_heartbeat.set()

        data, first = {"session": self.result_data}, True
        get_interval = lambda now: now + data["session"]["keep_method"]["heartbeat"]["lifetime"] / 1000 - 3
        make_url = lambda session_id: f"{URLS['base_heartbeat']}/{session_id}?_format=json&_method=PUT"
        after = get_interval(time())

        while not self._stop:
            now = time()
            # ここで定期的にHeartbeatを送ります。
            if now >= after:
                self.print("Sending heartbeat...", data)

                if first:
                    # 最初は普通とは違うやつもリクエストしないといけないのでする。
                    async with ClientSession(raise_for_status=True) as session:
                        async with session.options(
                            make_url(session_id), headers=self._headers[0]
                        ) as r:
                            r.raise_for_status()
                    first = False

                async with ClientSession(raise_for_status=True) as session:
                    async with session.post(
                        make_url(session_id),
                        headers=self._headers[1], json=data
                    ) as r:
                        self.result_data = (await r.json(loads=loads))["data"]["session"]

                self.print("Done.")
                data = {"session": self.result_data}

                after = get_interval(now)
            else:
                await asyncio.sleep(0.05)

Methods

def close(self) ‑> None

NicoNicoVideoAsyncを終了します。
動画のダウンロードに必要な通信をするHeartbeatを止めます。
もし動画のダウンロードが終わったのならこれを実行してください。
async with構文を使用するのならこれを実行する必要はありません。
async with構文の使用例はconnectにあります。

Warnings

これを使用してもイベントループは閉じません。

Expand source code

def close(self) -> None:
    """NicoNicoVideoAsyncを終了します。  
    動画のダウンロードに必要な通信をするHeartbeatを止めます。  
    もし動画のダウンロードが終わったのならこれを実行してください。  
    `async with`構文を使用するのならこれを実行する必要はありません。  
    `async with`構文の使用例は`connect`にあります。

    Warnings
    --------
    これを使用してもイベントループは閉じません。"""
    self._stop = True
    if self.heartbeat_task:
        try:
            self.heartbeat_task.cancel()
        except Exception as e:
            if not isinstance(e, asyncio.CancelledError):
                raise e

async def connect(self) ‑> None

Notes

これとcloseは以下のようにasync with構文で省略が可能です。

url = "https://www.nicovideo.jp/watch/sm9664372"
async with niconico_dl.NicoNicoVideoAsync(url) as nico:
    await nico.download("video.mp4")

Expand source code

async def connect(self) -> None:
    """ニコニコ動画に接続します。  
    `download`を使用するにはこれを実行してからする必要があります。  
    そしてニコニコ動画との通信が終了したのなら`close`を実行する必要があります。

    Notes
    -----
    これと`close`は以下のように`async with`構文で省略が可能です。  
    ```python
    url = "https://www.nicovideo.jp/watch/sm9664372"
    async with niconico_dl.NicoNicoVideoAsync(url) as nico:
        await nico.download("video.mp4")
    ```"""
    self.heartbeat_task = self.loop.create_task(
        self._heartbeat(), name="niconico_dl.heartbeat"
    )
    await self.wait_until_working_heartbeat()

async def download(self, path: str, load_chunk_size: int = 1024) ‑> None

ニコニコ動画の動画をダウンロードします。
mp4形式でダウンロードされます。

Examples

url = "https://www.nicovideo.jp/watch/sm9720246"
async with NicoNicoVideoAsync(url) as nico:
    data = await nico.get_info()
    title = data["video"]["title"]
    await nico.download(title + ".mp4")

Notes

もしasync with構文を使用しないでダウンロードする場合は、ダウンロード前にconnectを、ダウンロード終了後にcloseを実行してください。
動画データの通信に必要なHeartbeatが永遠に動き続けることになります。

Parameters

path : str: ダウンロードするニコニコ動画の動画の保存先です。
load_chunk_size : int, default 1024: 一度にどれほどの量をダウンロードするかです。

Expand source code

async def download(self, path: str, load_chunk_size: int = 1024) -> None:
    """ニコニコ動画の動画をダウンロードします。  
    mp4形式でダウンロードされます。

    Examples
    --------
    ```python
    url = "https://www.nicovideo.jp/watch/sm9720246"
    async with NicoNicoVideoAsync(url) as nico:
        data = await nico.get_info()
        title = data["video"]["title"]
        await nico.download(title + ".mp4")
    ```

    Notes
    -----
    もし`async with`構文を使用しないでダウンロードする場合は、ダウンロード前に`connect`を、ダウンロード終了後に`close`を実行してください。  
    動画データの通信に必要なHeartbeatが永遠に動き続けることになります。

    Parameters
    ----------
    path : str
        ダウンロードするニコニコ動画の動画の保存先です。
    load_chunk_size : int, default 1024
        一度にどれほどの量をダウンロードするかです。"""
    self.print("Now loading...")
    url = await self.get_download_link()

    params = (
        (
            "ht2_nicovideo",
            self.result_data["content_auth"]["content_auth_info"]["value"]
        ),
    )
    headers = self._headers[1]
    headers["Content-Type"] = "video/mp4"

    BASE = "Downloading video... :"
    self.print(BASE, "Now loading...", first="\r", end="")

    async with ClientSession(raise_for_status=True) as session:
        async with session.get(url, headers=headers, params=params) as r:
            size, now_size = r.content_length, 0

            self.print(BASE, "Making a null file...", first="\r", end="")

            async with async_open(path, "wb") as f:
                await f.write(b"")
            async with async_open(path, "ab") as f:
                async for chunk in r.content.iter_chunked(load_chunk_size):
                    if chunk:
                        now_size += len(chunk)
                        await f.write(chunk)
                        self.print(
                            BASE,
                            f"{int(now_size/size*100)}% ({now_size}/{size})",
                            first="\r", end=""
                        )

    self.print("Done.")

async def get_download_link(self) ‑> str

ニコニコ動画の動画のダウンロードリンクを取得します。
返されるリンクからはmp4の動画をダウンロードできます。

Returns

str : Download link

Warnings

返されるダウンロードリンクはHeartbeatが動いている間でしか使うことができません。
ですのでHeartbeatを止めるcloseはダウンロードリンクの使用が終わってから実行しましょう。
したがってasync with構文を使用して取得したダウンロードリンクはすぐに使えなくなることがあるので注意してください。

Expand source code

async def get_download_link(self) -> str:
    """
    ニコニコ動画の動画のダウンロードリンクを取得します。  
    返されるリンクからはmp4の動画をダウンロードできます。

    Returns
    -------
    str : Download link

    Warnings
    --------
    返されるダウンロードリンクはHeartbeatが動いている間でしか使うことができません。  
    ですのでHeartbeatを止める`close`はダウンロードリンクの使用が終わってから実行しましょう。  
    したがって`async with`構文を使用して取得したダウンロードリンクはすぐに使えなくなることがあるので注意してください。
    """
    if self._download_link is None:
        # Heartbeatが動いていないなら動かす。
        if not self.is_working_heartbeat():
            await self.connect()
        # Heartbeatが動画のURLを取得するまで待機する。
        await self.wait_until_working_heartbeat()
        self._download_link = self.result_data["content_uri"]

    return self._download_link

async def get_info(self) ‑> dict

ニコニコ動画のウェブページから動画のデータを取得するコルーチン関数です。

Returns

data : dict: 取得した動画の情報です。

Raises

NicoNicoAcquisitionFailed: ニコニコ動画から情報を取得するのに失敗した際に発生します。

Expand source code

async def get_info(self) -> dict:
    """ニコニコ動画のウェブページから動画のデータを取得するコルーチン関数です。

    Returns
    -------
    data : dict
        取得した動画の情報です。

    Raises
    ------
    NicoNicoAcquisitionFailed
        ニコニコ動画から情報を取得するのに失敗した際に発生します。"""
    self.print("Getting video data...")
    if not self._data:
        # もし動画データを取得していないなら動画URLのHTMLから動画データを取得する。
        # Heartbeatの通信にも必要なものでもあります。
        async with ClientSession(raise_for_status=True) as session:
            async with session.get(self._url, headers=self._headers[2]) as r:
                soup = BeautifulSoup(await r.text(), "html.parser")
                data = soup.find(
                    "div", {"id": "js-initial-watch-data"}
                ).get("data-api-data")
                if data:
                    self._data = loads(data)
                else:
                    raise NicoNicoAcquisitionFailed(
                        "ニコニコ動画から情報を取得するのに失敗しました。"
                    )
    self.print("Done.")
    return self._data

def is_working_heartbeat(self) ‑> bool

Heartbeatが動いているかの真偽値を返します。

Returns

is_working : bool: Heartbeatが動いているかどうかの真偽値です。

Expand source code

def is_working_heartbeat(self) -> bool:
    """Heartbeatが動いているかの真偽値を返します。

    Returns
    -------
    is_working : bool
        Heartbeatが動いているかどうかの真偽値です。"""
    return self._working_heartbeat.is_set()

def print(self, *args, first: str = '', **kwargs) ‑> None

niconico_dl用に用意したログ出力用のprintです。

Expand source code

def print(self, *args, first: str = "", **kwargs) -> None:
    """niconico_dl用に用意したログ出力用の`print`です。"""
    if self._log:
        print(
            f"{first}[niconico_dl_async]",
            *args, **kwargs
        )

async def wait_until_working_heartbeat(self) ‑> None

Heartbeatが動き出すまで待機します。

Expand source code

async def wait_until_working_heartbeat(self) -> None:
    """Heartbeatが動き出すまで待機します。"""
    await self._working_heartbeat.wait()

Package `niconico_dl`

niconico_dl

Install

Examples

Normal

Async

Command Line

Notes

Sub-modules

Classes

Ancestors

Parameters

Attributes

Seealso

Methods

Notes

Examples

Notes

Parameters

Returns

Warnings

Returns

Raises

Returns

Parameters

Attributes

Seealso

Methods

Warnings

Notes

Examples

Notes

Parameters

Returns

Warnings

Returns

Raises

Returns