bazarr/bazarr/subtitles/adaptive_searching.py

# coding=utf-8
# fmt: off

import ast
import logging

from datetime import datetime, timedelta

from app.config import settings


def is_search_active(desired_language, attempt_string):
    """
    Function to test if it's time to search again after a previous attempt matching the desired language. For 3 weeks,
    we search on a scheduled basis but after 3 weeks we start searching only once a week.

    @param desired_language: 2 letters language to search for in attempts
    @type desired_language: str
    @param attempt_string: string representation of a list of lists from database column failedAttempts
    @type attempt_string: str

    @return: return True if it's time to search again and False if not
    @rtype: bool
    """

    if settings.general.adaptive_searching:
        logging.debug("Adaptive searching is enable, we'll see if it's time to search again...")
        try:
            # let's try to get a list of lists from the string representation in database
            attempts = ast.literal_eval(attempt_string)
            if type(attempts) is not list:
                # attempts should be a list if not, it's malformed or None
                raise ValueError
        except ValueError:
            logging.debug("Adaptive searching: attempts is malformed. As a failsafe, search will run.")
            return True

        if not len(attempts):
            logging.debug("Adaptive searching: attempts list is empty, search will run.")
            return True

        # get attempts matching the desired language and sort them by timestamp ascending
        matching_attempts = sorted([x for x in attempts if x[0] == desired_language], key=lambda x: x[1])

        if not len(matching_attempts):
            logging.debug("Adaptive searching: there's no attempts matching desired language, search will run.")
            return True
        else:
            logging.debug(f"Adaptive searching: attempts matching language {desired_language}: {matching_attempts}")

        # try to get the initial and latest search timestamp from matching attempts
        initial_search_attempt = matching_attempts[0]
        latest_search_attempt = matching_attempts[-1]

        # try to parse the timestamps for those attempts
        try:
            initial_search_timestamp = datetime.fromtimestamp(initial_search_attempt[1])
            latest_search_timestamp = datetime.fromtimestamp(latest_search_attempt[1])
        except (OverflowError, ValueError, OSError):
            logging.debug("Adaptive searching: unable to parse initial and latest search timestamps, search will run.")
            return True
        else:
            logging.debug(f"Adaptive searching: initial search date for {desired_language} is "
                          f"{initial_search_timestamp}")
            logging.debug(f"Adaptive searching: latest search date for {desired_language} is {latest_search_timestamp}")

        # defining basic calculation variables
        now = datetime.now()
        if settings.general.adaptive_searching_delay.endswith('d'):
            extended_search_delay = timedelta(days=int(settings.general.adaptive_searching_delay[:1]))
        elif settings.general.adaptive_searching_delay.endswith('w'):
            extended_search_delay = timedelta(weeks=int(settings.general.adaptive_searching_delay[:1]))
        else:
            logging.debug(f"Adaptive searching: cannot parse adaptive_searching_delay from config file: "
                          f"{settings.general.adaptive_searching_delay}")
            return True
        logging.debug(f"Adaptive searching: delay after initial search value: {extended_search_delay}")

        if settings.general.adaptive_searching_delta.endswith('d'):
            extended_search_delta = timedelta(days=int(settings.general.adaptive_searching_delta[:1]))
        elif settings.general.adaptive_searching_delta.endswith('w'):
            extended_search_delta = timedelta(weeks=int(settings.general.adaptive_searching_delta[:1]))
        else:
            logging.debug(f"Adaptive searching: cannot parse adaptive_searching_delta from config file: "
                          f"{settings.general.adaptive_searching_delta}")
            return True
        logging.debug(f"Adaptive searching: delta between latest search and now value: {extended_search_delta}")

        if initial_search_timestamp + extended_search_delay > now:
            logging.debug(f"Adaptive searching: it's been less than {settings.general.adaptive_searching_delay} since "
                          f"initial search, search will run.")
            return True
        else:
            logging.debug(f"Adaptive searching: it's been more than {settings.general.adaptive_searching_delay} since "
                          f"initial search, let's check if it's time to search again.")
            if latest_search_timestamp + extended_search_delta <= now:
                logging.debug(
                    f"Adaptive searching: it's been more than {settings.general.adaptive_searching_delta} since "
                    f"latest search, search will run.")
                return True
            else:
                logging.debug(
                    f"Adaptive searching: it's been less than {settings.general.adaptive_searching_delta} since "
                    f"latest search, we're not ready to search yet.")
                return False

    logging.debug("adaptive searching is disabled, search will run.")
    return True


def updateFailedAttempts(desired_language, attempt_string):
    """
    Function to parse attempts and make sure we only keep initial and latest search timestamp for each language.

    @param desired_language: 2 letters language to search for in attempts
    @type desired_language: str
    @param attempt_string: string representation of a list of lists from database column failedAttempts
    @type attempt_string: str

    @return: return a string representation of a list of lists like [str(language_code), str(attempts)]
    @rtype: str
    """

    try:
        # let's try to get a list of lists from the string representation in database
        attempts = ast.literal_eval(attempt_string)
        logging.debug(f"Adaptive searching: current attempts value is {attempts}")
        if type(attempts) is not list:
            # attempts should be a list if not, it's malformed or None
            raise ValueError
    except ValueError:
        logging.debug("Adaptive searching: failed to parse attempts value, we'll use an empty list.")
        attempts = []

    matching_attempts = sorted([x for x in attempts if x[0] == desired_language], key=lambda x: x[1])
    logging.debug(f"Adaptive searching: attempts matching language {desired_language}: {matching_attempts}")

    filtered_attempts = sorted([x for x in attempts if x[0] != desired_language], key=lambda x: x[1])
    logging.debug(f"Adaptive searching: attempts not matching language {desired_language}: {filtered_attempts}")

    # get the initial search from attempts if there's one
    if len(matching_attempts):
        filtered_attempts.append(matching_attempts[0])

    # append current attempt with language and timestamp to attempts
    filtered_attempts.append([desired_language, datetime.timestamp(datetime.now())])

    updated_attempts = sorted(filtered_attempts, key=lambda x: x[0])
    logging.debug(f"Adaptive searching: updated attempts that will be saved to database is {updated_attempts}")

    return str(updated_attempts)