delarte/src/delarte/__main__.py

# License: GNU AGPL v3: http://www.gnu.org/licenses/
# This file is part of `delarte` (https://git.afpy.org/fcode/delarte.git)

"""delarte - ArteTV downloader.

Usage:
  delarte (-h | --help)
  delarte --version
  delarte [options] URL
  delarte [options] URL RENDITION
  delarte [options] URL RENDITION VARIANT

Download a video from ArteTV streaming service. Omit RENDITION and/or
VARIANT to print the list of available values.

Arguments:
  URL         the URL from ArteTV website
  RENDITION   the rendition code [audio/subtitles language combination]
  VARIANT     the variant code [video quality version]

Options:
  -h --help              print this message
  --version              print current version of the program
  --debug                on error, print debugging information
  --name-use-id          use the program ID
  --name-sep=<sep>       field separator [default:  - ]
  --name-seq-pfx=<pfx>   sequence counter prefix [default:  - ]
  --name-seq-no-pad      disable sequence zero-padding
  --name-add-rendition   add rendition code
  --name-add-variant     add variant code
"""

import itertools
import sys
import time

import docopt
import urllib3

from . import (
    ModuleError,
    UnexpectedError,
    HTTPError,
    __version__,
    download_targets,
    fetch_program_sources,
    fetch_rendition_sources,
    fetch_targets,
    fetch_variant_sources,
)


class Abort(ModuleError):
    """Aborted."""


class Fail(UnexpectedError):
    """Unexpected error."""


def _create_progress():
    # create a progress handler for input downloads
    state = {}

    def on_progress(file, current, total):
        now = time.time()

        if current == 0:
            print(f"Downloading {file!r}: 0.0%", end="")
            state["start_time"] = now
            state["last_time"] = now
            state["last_count"] = 0

        elif current == total:
            elapsed_time = now - state["start_time"]
            rate = int(total / elapsed_time) if elapsed_time else "NaN"
            print(f"\rDownloading {file!r}: 100.0% [{rate}]")
            state.clear()

        elif now - state["last_time"] > 1:
            elapsed_time1 = now - state["start_time"]
            elapsed_time2 = now - state["last_time"]
            progress = int(1000.0 * current / total) / 10.0
            rate1 = int(current / elapsed_time1) if elapsed_time1 else "NaN"
            rate2 = (
                int((current - state["last_count"]) / elapsed_time2)
                if elapsed_time2
                else "NaN"
            )
            print(
                f"\rDownloading {file!r}: {progress}% [{rate1}, {rate2}]",
                end="",
            )
            state["last_time"] = now
            state["last_count"] = current

    return on_progress


def _select_rendition_sources(rendition_code, rendition_sources):
    if rendition_code:
        filtered = [s for s in rendition_sources if s.rendition.code == rendition_code]
        if filtered:
            return filtered
        print(
            f"{rendition_code!r} is not a valid rendition code. Available values are:"
        )
    else:
        print("Available renditions:")

    key = lambda s: (s.rendition.label, s.rendition.code)

    rendition_sources.sort(key=key)
    for (label, code), _ in itertools.groupby(rendition_sources, key=key):
        print(f"{code:>12} : {label}")

    raise Abort()


def _select_variant_sources(variant_code, variant_sources):
    if variant_code:
        filtered = [s for s in variant_sources if s.variant.code == variant_code]
        if filtered:
            return filtered
        print(f"{variant_code!r} is not a valid variant code. Available values are:")
    else:
        print("Available variants:")

    variant_sources.sort(key=lambda s: s.video_media.track.height, reverse=True)
    for code, _ in itertools.groupby(variant_sources, key=lambda s: s.variant.code):
        print(f"{code:>12}")

    raise Abort()


def main():
    """CLI command."""
    args = docopt.docopt(__doc__, sys.argv[1:], version=__version__)

    http = urllib3.PoolManager(timeout=5)

    try:
        program_sources = fetch_program_sources(args["URL"], http)

        rendition_sources = _select_rendition_sources(
            args["RENDITION"],
            fetch_rendition_sources(program_sources, http),
        )

        variant_sources = _select_variant_sources(
            args["VARIANT"],
            fetch_variant_sources(rendition_sources, http),
        )

        targets = fetch_targets(
            variant_sources,
            http,
            **{
                k[7:].replace("-", "_"): v
                for k, v in args.items()
                if k.startswith("--name-")
            },
        )

        download_targets(targets, http, _create_progress())

    except UnexpectedError as e:
        if args["--debug"]:
            raise e
        print(str(e))
        print()
        print(
            "This program is the result of browser/server traffic analysis and involves\n"
            "some level of trying and guessing. This error might mean that we did not try\n"
            "enough or that we guessed poorly."
        )
        print("")
        print("Please consider submitting the issue to us so we may fix it.")
        print("")
        print("Issue tracker: https://git.afpy.org/fcode/delarte/issues")
        print(f"Title: {e.args[0]}")
        print("Body:")
        print(f"    {repr(e)}")
        return 1

    except ModuleError as e:
        if args["--debug"]:
            raise e
        print(str(e))
        return 1

    except HTTPError as e:
        if args["--debug"]:
            raise e
        print("Network error.")
        return 1


if __name__ == "__main__":
    sys.exit(main())