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-use-slug        use the URL slug
  --name-sep=<sep>       field separator [default:  - ]
  --name-seq-pfx=<pfx>   sequence counter prefix [default:  - ]
  --name-seq-no-pad      disable sequence zero-padding
  --name-add-resolution  add resolution tag
"""

import sys
import time

import docopt
import requests

from . import (
    __version__,
    compile_sources,
    download_target,
    fetch_sources,
    iter_renditions,
    iter_variants,
    select_rendition,
    select_variant,
)
from .error import ModuleError, UnexpectedError


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


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


_LANGUAGES = {
    "de": "German",
    "en": "English",
    "es": "Spanish",
    "fr": "French",
    "it": "Italian",
    "mul": "multiple language",
    "no": "Norwegian",
    "pt": "Portuguese",
}


def _language_name_for_code(code):
    return _LANGUAGES.get(code, f"[{code}]")


def _language_name(meta):
    return _language_name_for_code(meta.language)


def _print_renditions(renditions):
    has_original = False
    for code, (a_meta, s_meta) in renditions:
        label = _language_name(a_meta)
        if a_meta.is_original:
            has_original = True
            label = "original " + label
        elif a_meta.is_descriptive:
            label += " audio description"
        elif has_original:
            label += " dubbed"

        if s_meta:
            if s_meta.is_descriptive:
                label += f" ({_language_name(s_meta)} closed captions)"
            elif s_meta.language != a_meta.language:
                label += f" ({_language_name(s_meta)} subtitles)"

        print(f"\t{code:>6} - {label}")


def _validate_rendition(renditions, code):
    for code_, rendition in renditions:
        if code_ == code:
            break
    else:
        print(f"{code!r} is not a valid rendition code, possible values are:")
        _print_renditions(renditions)
        raise Abort()

    return rendition


def _print_variants(variants):
    for code, _ in variants:
        print(f"\t{code}")


def _validate_variant(variants, code):
    for code_, variant in variants:
        if code_ == code:
            break
    else:
        print(f"{code!r} is not a valid variant code, possible values are:")
        _print_variants(variants)
        raise Abort()

    return variant


def create_progress():
    """Create a progress handler for input downloads."""
    state = {
        "last_update_time": 0,
        "last_channel": None,
    }

    def progress(channel, current, total):
        now = time.time()

        if current == total:
            print(f"\rDownloading {channel}: 100.0%")
            state["last_update_time"] = now
        elif channel != state["last_channel"]:
            print(f"Downloading {channel}: 0.0%", end="")
            state["last_update_time"] = now
            state["last_channel"] = channel
        elif now - state["last_update_time"] > 1:
            print(
                f"\rDownloading {channel}: {int(1000.0 * current / total) / 10.0}%",
                end="",
            )
            state["last_update_time"] = now

    return progress


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

    try:

        http_session = requests.sessions.Session()

        sources = fetch_sources(http_session, args["URL"])

        renditions = list(iter_renditions(sources))
        if not args["RENDITION"]:
            print(f"Available renditions:")
            _print_renditions(renditions)
            return 0

        select_rendition(sources, _validate_rendition(renditions, args["RENDITION"]))

        variants = list(iter_variants(sources))
        if not args["VARIANT"]:
            print(f"Available variants:")
            _print_variants(variants)
            return 0

        select_variant(sources, _validate_variant(variants, args["VARIANT"]))

        target = compile_sources(
            sources,
            **{
                k[7:].replace("-", "_"): v
                for k, v in args.items()
                if k.startswith("--name-")
            },
        )

        progress = create_progress()

        download_target(http_session, target, progress)

    except UnexpectedError as 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:
        print(str(e))
        if args["--debug"]:
            print(repr(e))
        return 1

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


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