API¶

@validate_call
def calculate_consumption_time(
    html: FilePath,
    words_per_minute: NonNegativeInt = 265,
    multimedia_duration_resolver=None,  # noqa: ANN001 (unsuppored by Typer)
) -> int:
    """Calculate the consumption time of an HTML file in seconds.

    Uses concurrency to get the duration of any multimedia in the file to avoid any
    possible throttling.

    Args:
        html: Path to the HTML file whose consumption time will be calculated.
        words_per_minute: Reading speed in words per minute.
        multimedia_duration_resolver: Function used to get the duration of a
            multimedia file.

    Returns:
        The time in seconds to consume the content of the HTML file.
    """
    # This is the default multimedia duration resolver. It can't be set as the
    # default in the function parameters because it seems like you can't reuse
    # function parameters in Python.
    if multimedia_duration_resolver is None:

        def multimedia_duration_resolver(src: str) -> int:
            return get_multimedia_duration(html, src)

    raw_html: str = html.read_text("utf-8")
    soup: BeautifulSoup = BeautifulSoup(raw_html, "lxml")
    text: str | None = trafilatura.extract(raw_html)
    word_count, cjk_character_count = get_word_count(text or "")
    reading_time: int = calculate_reading_time(
        word_count, cjk_character_count, words_per_minute
    )

    image_count: int = len(soup("img"))
    image_time: int = calculate_viewing_time(image_count)

    multimedias: list[str] = extract_multimedias(soup)
    multimedia_time: int = 0

    with ThreadPoolExecutor() as e:
        resolved_durations: Iterator[int] = e.map(
            multimedia_duration_resolver, multimedias
        )

        multimedia_time += sum(resolved_durations)

    multimedia_time += get_custom_player_duration(html)

    return reading_time + image_time + multimedia_time

@validate_call
def calculate_consumption_time(
    container: FilePath, words_per_minute: NonNegativeInt = 265
) -> int:
    """Calculate the consumption time of a text container file in seconds.

    Args:
        container: Path to a file primarily meant for text. Supported types
            are EPUB, MOBI and PDF.
        words_per_minute: Reading speed in words per minute.

    Returns:
        The time in seconds to consume the content of the file.
    """
    text: str = extract_text(container)
    word_count, cjk_character_count = get_word_count(text)
    reading_time: int = calculate_reading_time(
        word_count, cjk_character_count, words_per_minute
    )

    return reading_time

@validate_call
def calculate_reading_time(
    word_count: int, cjk_character_count: int, words_per_minute: NonNegativeInt = 265
) -> int:
    """Calculate the reading time in seconds based on word count.

    Supports Chinese, Japanese, and Korean (CJK) by having its reading speed as
    1.8867924528 (500 / 265) that of the word one. This is done because, in the Medium
    formula, the average reading speed for words is 265 per minute, while the
    average for non-alphabetical languages is 500 per character.

    Args:
        word_count: The number of words in the text.
        cjk_character_count: The number of CJK characters in the text.
        words_per_minute: Reading speed in words per minute.

    Returns:
        How long in seconds it would take to read the text.
    """
    minutes_to_seconds: int = 60
    word_reading_time: int | float = (
        word_count / words_per_minute
    ) * minutes_to_seconds
    # 500 / 265 \approx 1.8867924528301887.
    cjk_reading_rate: int | float = words_per_minute * 1.8867924528301887
    cjk_reading_time: int | float = (
        cjk_character_count / cjk_reading_rate
    ) * minutes_to_seconds
    raw_reading_time: int | float = word_reading_time + cjk_reading_time
    reading_time: int = math.ceil(raw_reading_time)

    return reading_time

def calculate_consumption_time(
    container: FilePath, words_per_minute: NonNegativeInt = 265
) -> int:
    """Calculate the consumption time of a plain text file in seconds.

    Args:
        container: Path to the plain text file whose consumption time will be
            calculated.
        words_per_minute: Reading speed in words per minute.

    Returns:
        The time in seconds to consume the content in the plain text file.
    """
    text: str = container.read_text("utf-8")
    word_count, cjk_character_count = get_word_count(text)

    return calculate_reading_time(word_count, cjk_character_count, words_per_minute)

@validate_call
def calculate_consumption_time(
    url: HttpUrl, words_per_minute: NonNegativeInt = 265
) -> int:
    """Calculate the consumption time of a URL in seconds.

    Avoids code duplication by downloading the HTML of the URL to a temporary
    file, to use the HTML backend [`calculate_html_consumption_time`][consumo.calculate_html_consumption_time].

    Args:
        url: URL pointing to the content whose consumption time will be analyzed.
        words_per_minute: Reading speed in words per minute.

    Returns:
        The time in seconds to consume the content the URL points to.

    Raises:
        ConnectionError: When the HTML content of the URL wasn't downloaded.
    """
    html_content: str | None = trafilatura.fetch_url(str(url))

    if html_content is None:
        raise ConnectionError("No network connection or program blocked by Cloudflare")

    with TemporaryDirectory() as tmp_dir:
        html: Path = Path(tmp_dir) / "temp.html"

        html.write_text(html_content, "utf-8")

        return calculate_html_consumption_time(
            html, words_per_minute, lambda src: get_multimedia_duration(url, src)
        )

@validate_call
def calculate_viewing_time(image_count: NonNegativeInt) -> int:
    """Calculate the time for viewing images based on count.

    Args:
        image_count: The number of images.

    Returns:
        The time in seconds to view all the images.
    """
    first_ten_images_time: int = 75

    if image_count > 10:
        after_ten_images_time: int = (image_count - 10) * 3

        return first_ten_images_time + after_ten_images_time

    # For the first ten images, the time follows an arithmetic progression of:
    # S_{n} = \frac{n(25 - n)}{2}
    image_time: int | float = (image_count * (25 - image_count)) / 2

    return math.ceil(image_time)

@validate_call
def extract_text(container: FilePath) -> str:
    """Extract text from a text container file.

    Args:
        container: Path to a file primarily meant for text. Supported types
            are EPUB, MOBI and PDF.

    Returns:
        All the text content in the container.
    """
    with pymupdf.open(container) as c:
        raw_text: str = " ".join(page.get_text() for page in c)

        return raw_text.strip()

def extract_multimedias(soup: BeautifulSoup) -> list[str]:
    """Get all the multimedia sources from an HTML file.

    Args:
        soup: The HTML file as parsed by BeautifulSoup.

    Returns:
        A list of all the multimedia sources.
    """
    audios: ResultSet[Tag] = soup("audio")
    iframes: ResultSet[Tag] = soup("iframe")
    videos: ResultSet[Tag] = soup("video")
    result: list[str] = []

    for audio in audios:
        primary_source: Tag = audio("source")[0]
        src: str | AttributeValueList | None = primary_source.get("src")

        result.append(str(src))

    for iframe in iframes:
        src: str | AttributeValueList | None = iframe.get("src")

        result.append(str(src))

    for video in videos:
        primary_source: Tag = video("source")[0]
        src: str | AttributeValueList | None = primary_source.get("src")

        result.append(str(src))

    return result

def format_time(total_seconds: int) -> str:
    """Format the duration/consumption time given in seconds in a *h *m *s format.

    Args:
        total_seconds: The duration/consumption time in seconds of the content.

    Returns:
        The duration/consumption time in a *h *m *s format.
    """
    minutes, seconds = divmod(total_seconds, 60)
    hours, minutes = divmod(minutes, 60)
    hours %= 24

    parts: list[str] = []

    if hours:
        parts.append(f"{hours}h")
    if minutes:
        parts.append(f"{minutes}m")

    parts.append(f"{seconds}s")

    return " ".join(parts)

@validate_call
def get_custom_player_duration(html: FilePath) -> int:
    """Parse the JSON data in an HTML file provided for SEO to get video duration.

    Designed with videos using custom players like the BBC's smp-toucan-player
    in mind.

    The supported format for duration is ISO 8601.

    Args:
        html: Path to the HTML file whose content will be parsed for JSON data
            containing duration information.

    Returns:
        The duration reported by the JSON data as an integer representing
        seconds.
    """
    raw_html: str = html.read_text("utf-8")
    soup: BeautifulSoup = BeautifulSoup(raw_html, "lxml")
    script_attrs = {"data-schema": "video-object"}

    found_script_tags: ResultSet[Tag] = soup("script", script_attrs)
    total_seconds: int = 0

    for tag in found_script_tags:
        if not tag.string:
            continue

        try:
            data: Any = json.loads(tag.string)
            duration_str: str | None = data.get("duration")

            if duration_str:
                duration: timedelta = DURATION_ADAPTER.validate_python(duration_str)

                total_seconds += int(duration.total_seconds())

        except Exception:
            continue

    return total_seconds

@validate_call
def get_duration(
    file: FilePath,
    words_per_minute: NonNegativeInt = 265,
) -> int:
    """Get the duration or calculate the consumption time of a file in seconds.

    Support is based on MIME type.

    Supported types are:

    - "audio": [`get_multimedia_duration`][consumo.get_multimedia_duration].
    - "image": [`calculate_viewing_time`][consumo.calculate_viewing_time].
    - "video": [`get_multimedia_duration`][consumo.get_multimedia_duration].

    Supported types/subtypes are:

    - "application/epub+zip": [`calculate_mass_media_consumption_time`][consumo.calculate_mass_media_consumption_time].
    - "application/pdf": [`calculate_mass_media_consumption_time`][consumo.calculate_mass_media_consumption_time].
    - "application/x-mobipocket-ebook": [`calculate_mass_media_consumption_time`][consumo.calculate_mass_media_consumption_time].
    - "text/html": [`calculate_html_consumption_time`][consumo.calculate_html_consumption_time].
    - "text/plain": [`calculate_text_consumption_time`][consumo.calculate_text_consumption_time].

    Directories are unsupported.

    Args:
        file: The path to the file whose duration or consumption time will be
            analyzed.
        words_per_minute: Reading speed in words per minute.

    Returns:
        The time in seconds to consume the content in the file.

    Raises:
        UnsupportedMIMETypeError: When the MIME type is unsupported.
    """
    mime_type: str = magic.from_file(str(file), mime=True)
    type, subtype = mime_type.split("/", 1)
    multimedia_types = ("audio", "video")

    if type == "image":
        return calculate_viewing_time(1)

    if type in multimedia_types:
        return get_multimedia_duration(file)

    mime_type_handler: dict[str, Callable[[Path, NonNegativeInt], int]] = {
        "application/epub+zip": calculate_mass_media_consumption_time,
        "application/pdf": calculate_mass_media_consumption_time,
        "application/x-mobipocket-ebook": calculate_mass_media_consumption_time,
        "text/html": calculate_html_consumption_time,
        "text/plain": calculate_text_consumption_time,
    }

    handler: Callable[[Path, NonNegativeInt], int] | None = mime_type_handler.get(
        mime_type
    )

    if handler is None:
        raise UnsupportedMIMETypeError("File type not supported")

    return handler(file, words_per_minute)

@validate_call
def get_multimedia_duration(html: FilePath, src: str) -> int:
    """Get the duration of a multimedia file in an HTML file.

    Tries to treat the multimedia file as if it was hosted online, then tries to
    resolve its path if that fails.

    Args:
        html: Path to the HTML file where the multimedia file was found.
        src: Path used for the file's "src" attribute.

    Returns:
        The duration of the content in seconds.
    """
    try:
        return get_absolute_path_multimedia_duration(HttpUrl(src))
    except ValidationError:
        # If HttpUrl(src) fails validation, the src is likely a relative
        # path rather than a URL.
        return get_relative_path_multimedia_duration(html, Path(src))

@validate_call
def get_multimedia_duration(container: FilePath | HttpUrl) -> int:
    """Get the duration from a multimedia container or URL.

    Args:
        container: Either a path to a multimedia container or a URL.

    Returns:
        The duration in seconds of the content.

    Raises:
        MissingMetadataError: If the duration can't be found from the metadata.
    """
    av_options: dict[str, str] = {
        # Disables duration analysis, as getting the duration from metadata is
        # faster.
        "analyzeduration": "0",
        # Allocate enough space to probe an integer (32 bits), which will be our
        # duration.
        "probesize": "32",
    }

    with av.open(str(container), options=av_options) as c:
        duration: int | None = c.duration

        if duration is None:
            raise MissingMetadataError("duration not found")

        # PyAV stores the duration as an integer with more precision than we
        # need (for example, 1 second is equivalent to 1000000).
        # PyAV provides the time_base value to correct this.
        raw_seconds: float = duration / av.time_base

        return math.ceil(raw_seconds)

@validate_call
def get_duration(
    url: HttpUrl,
    words_per_minute: NonNegativeInt = 265,
    depth: NonNegativeInt = 0,
    cache: bool = True,
    cache_dir: Path = Path.cwd(),
    get_cached_resolver: Callable[[Path, str], int] = dummy_get_cached_resolver,
    cache_resolver: Callable[[Path, str, int, int], None] = dummy_cache_resolver,
) -> int:
    """Get the duration or calculate the consumption time of a URL in seconds.

    Gets the duration of media from hosting platforms or direct file
    links, and calculates the consumption time otherwise.

    Args:
        url: The URL of the content whose duration or consumption time will be
            analyzed.
        words_per_minute: Reading speed in words per minute.
        depth: How many levels to recursively follow URLs on the page.
        cache: Whether to cache results in a database for later reuse.
            Values are invalidated based on time.
        cache_dir: The path to where the cache will be stored.
        get_cached_resolver: Function for getting a value from a cache system
            whose signature consists of cache directory, key, and time (date) for
            cache invalidation.
        cache_resolver: Function for storing a value in a cache system  whose
            signature consists of cache directory, key, value, and time (date) for
            cache invalidation.

    !!! warning

        `get_cached_resolver` and `cache_resolver` have dummy default values. You have
        to implement your own cache functions if you want to use cache!

    Returns:
        The time in seconds to consume the content the URL points to.
    """
    # 1 day in seconds.
    time_to_live: int = 86400
    normalized_url: HttpUrl = clean_url(url)
    key: str = f"{normalized_url.unicode_string()}:{words_per_minute}:{depth}"

    if cache:
        try:
            return get_cached_resolver(cache_dir, key)
        except NoCacheError:
            pass

    if is_hosted(url):
        result: int = get_hosted_multimedia_duration(url)

        if cache:
            cache_resolver(cache_dir, key, result, time_to_live)

        return result

    try:
        result: int = get_multimedia_duration(url)
    except FFmpegError:
        result: int = calculate_consumption_time(url, words_per_minute)

    if cache:
        cache_resolver(cache_dir, key, result, time_to_live)

    if depth > 0:
        with urllib.request.urlopen(str(url)) as response:
            raw_html: str = response.read()

        soup: BeautifulSoup = BeautifulSoup(raw_html, "lxml")
        tags: ResultSet[Tag] = soup("a")

        def recursive(tag: Tag) -> int:
            href: str | AttributeValueList | None = tag.get("href")

            absolute_url: HttpUrl = HttpUrl(urllib.parse.urljoin(str(url), href))

            return get_duration(
                absolute_url, words_per_minute=words_per_minute, depth=depth - 1
            )

        result += sum(map(recursive, tags))

    return result

def get_duration(url: HttpUrl) -> int:
    """Get the duration of a multimedia container hosted online.

    Args:
        url: URL pointing to where the multimedia container is hosted.

    Returns:
        The duration of the content in seconds.
    """
    if is_hosted(url):
        return get_hosted_multimedia_duration(url)

    return get_multimedia_duration(url)

def get_word_count(text: str) -> tuple[int, int]:
    """Get the number of words from text.

    Args:
        text: Text where the number of words will be counted from.

    Returns:
        The number of words in the text.
    """
    word_count: float = len(text.split())

    # \u4e00-\u9fff = CJK Unified Ideographs (Han).
    # \u3040-\u309f = Hiragana.
    # \u30a0-\u30ff = Katakana.
    # \uac00-\ud7af = Hangul Syllables.
    cjk: Pattern[str] = re.compile(
        r"[\u4e00-\u9fff\u3040-\u309f\u30a0-\u30ff\uac00-\ud7af]"
    )
    cjk_character_count: int = len(cjk.findall(text))

    return word_count, cjk_character_count