CastagnaIT/plugin.video.netflix

View on GitHub
resources/lib/common/device_utils.py

Summary

Maintainability
A
55 mins
Test Coverage
# -*- coding: utf-8 -*-
"""
    Copyright (C) 2017 Sebastian Golasch (plugin.video.netflix)
    Copyright (C) 2020 Stefano Gottardo - @CastagnaIT (original implementation module)
    Miscellaneous utility functions related to the device

    SPDX-License-Identifier: MIT
    See LICENSES/MIT.md for more information.
"""
import xbmc

from resources.lib.globals import G
from resources.lib.utils.esn import WidevineForceSecLev
from resources.lib.utils.logging import LOG


def select_port(service):
    """Select an unused port on the host machine for a server and store it in the settings"""
    port = select_unused_port()
    G.LOCAL_DB.set_value(f'{service.lower()}_service_port', port)
    LOG.info('[{}] Picked Port: {}', service, port)
    return port


def select_unused_port():
    """
    Helper function to select an unused port on the host machine

    :return: int - Free port
    """
    import socket
    from contextlib import closing
    # pylint: disable=no-member
    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
        sock.bind(('127.0.0.1', 0))
        _, port = sock.getsockname()
        return port


def get_system_platform():
    if not hasattr(get_system_platform, 'cached'):
        platform = "unknown"
        if xbmc.getCondVisibility('system.platform.linux') and not xbmc.getCondVisibility('system.platform.android'):
            platform = "linux"
        elif xbmc.getCondVisibility('system.platform.linux') and xbmc.getCondVisibility('system.platform.android'):
            platform = "android"
        elif xbmc.getCondVisibility('system.platform.uwp'):
            platform = "uwp"
        elif xbmc.getCondVisibility('system.platform.windows'):
            platform = "windows"
        elif xbmc.getCondVisibility('system.platform.osx'):
            platform = "osx"
        elif xbmc.getCondVisibility('system.platform.ios'):
            platform = "ios"
        elif xbmc.getCondVisibility('system.platform.tvos'):
            platform = "tvos"
        get_system_platform.cached = platform
    return get_system_platform.cached


def get_machine():
    """Get machine architecture"""
    from platform import machine
    try:
        return machine()
    except Exception:  # pylint: disable=broad-except
        # Due to OS restrictions on 'ios' and 'tvos' this generate an exception
        # See python limits in the wiki development page
        # Fallback with a generic arm
        return 'arm'


def is_android_tv(props=None):
    """Check Android properties to determine if the device has an Android TV system"""
    if props is None:
        props = get_android_system_props()
    ret = 'TV' in props.get('ro.build.characteristics', '').upper()
    # Xiaomi box devices like Mi Box 3, Mi Box S, Tv Stick, and maybe other models
    # don't have "tv" value into ro.build.characteristics
    # therefore we check the name from ro.com.google.clientidbase, that at least to the mentioned models are the same
    if (not ret and props.get('ro.product.manufacturer', '').upper() == 'XIAOMI'
        and props.get('ro.com.google.clientidbase', '').upper() == 'ANDROID-XIAOMI-TV'):
        ret = True
    return ret


def is_device_4k_capable():
    """Check if the device is 4k capable"""
    # Currently only on android is it possible to use 4K
    if get_system_platform() == 'android':
        from resources.lib.database.db_utils import TABLE_SESSION
        # Check if the drm has security level L1
        wv_force_sec_lev = G.LOCAL_DB.get_value('widevine_force_seclev',
                                                WidevineForceSecLev.DISABLED,
                                                table=TABLE_SESSION)
        is_l3_forced = wv_force_sec_lev != WidevineForceSecLev.DISABLED
        is_drm_l1_security_level = (G.LOCAL_DB.get_value('drm_security_level', '', table=TABLE_SESSION) == 'L1'
                                    and not is_l3_forced)
        # Check if HDCP level is 2.2 or up
        hdcp_4k_capable = get_hdcp_level() >= 2.2
        return bool(is_drm_l1_security_level and hdcp_4k_capable)
    return False


def is_device_l1_enabled():
    """Check if L1 security level is enabled"""
    from resources.lib.database.db_utils import TABLE_SESSION
    wv_force_sec_lev = G.LOCAL_DB.get_value('widevine_force_seclev',
                                            WidevineForceSecLev.DISABLED,
                                            table=TABLE_SESSION)
    is_l3_forced = wv_force_sec_lev != WidevineForceSecLev.DISABLED
    return G.LOCAL_DB.get_value('drm_security_level', '', table=TABLE_SESSION) == 'L1' and not is_l3_forced


def get_hdcp_level():
    """Get the HDCP level"""
    from re import findall
    from resources.lib.database.db_utils import TABLE_SESSION
    drm_hdcp_level = findall('\\d+\\.\\d+', G.LOCAL_DB.get_value('drm_hdcp_level', '', table=TABLE_SESSION))
    return float(drm_hdcp_level[0]) if drm_hdcp_level else 1.4


def get_user_agent(enable_android_mediaflag_fix=False):
    """
    Determines the user agent string for the current platform.
    Needed to retrieve a valid ESN (except for Android, where the ESN can be generated locally)

    :returns: str -- User agent string
    """
    system = get_system_platform()
    if enable_android_mediaflag_fix and system == 'android' and is_device_4k_capable():
        # The UA affects not only the ESNs in the login, but also the video details,
        # so the UAs seem refer to exactly to these conditions: https://help.netflix.com/en/node/23742
        # This workaround is needed because currently we do not login through the netflix native android API,
        # but redirect everything through the website APIs, and the website APIs do not really support android.
        # Then on android usually we use the 'arm' UA which refers to chrome os, but this is limited to 1080P, so the
        # labels on the 4K devices appears wrong (in the Kodi skin the 4K videos have 1080P media flags instead of 4K),
        # the Windows UA is not limited, so we can use it to get the right video media flags.
        system = 'windows'

    chrome_version = 'Chrome/108.0.0.0'
    base = 'Mozilla/5.0 '
    base += '%PL% '
    base += 'AppleWebKit/537.36 (KHTML, like Gecko) '
    base += '%CH_VER% Safari/537.36'.replace('%CH_VER%', chrome_version)

    if system in ['osx', 'ios', 'tvos']:
        return base.replace('%PL%', '(Macintosh; Intel Mac OS X 10_15_5)')
    if system in ['windows', 'uwp']:
        return base.replace('%PL%', '(Windows NT 10.0; Win64; x64)')
    # ARM based Linux
    machine_arch = get_machine()
    if machine_arch.startswith('arm'):
        # Last number is the platform version of Chrome OS
        return base.replace('%PL%', '(X11; CrOS armv7l 15183.69.0)')
    if machine_arch.startswith('aarch'):
        # Last number is the platform version of Chrome OS
        return base.replace('%PL%', '(X11; CrOS aarch64 15183.69.0)')
    # x86 Linux
    return base.replace('%PL%', '(X11; Linux x86_64)')


def is_internet_connected():
    """
    Check internet status
    :return: True if connected
    """
    if not xbmc.getCondVisibility('System.InternetState'):
        # Double check when Kodi say that it is not connected
        # i'm not sure the InfoLabel will work properly when Kodi was started a few seconds ago
        # using getInfoLabel instead of getCondVisibility often return delayed results..
        return _check_internet()
    return True


def _check_internet():
    """
    Checks via socket if the internet works (in about 0,7sec with no timeout error)
    :return: True if connected
    """
    import socket
    for timeout in [1, 1]:
        try:
            socket.setdefaulttimeout(timeout)
            host = socket.gethostbyname("www.google.com")
            s = socket.create_connection((host, 80), timeout)
            s.close()
            return True
        except Exception:  # pylint: disable=broad-except
            # Error when is not reachable
            pass
    return False


def get_supported_hdr_types():
    """
    Get supported HDR types by the display
    :return: supported type as list ['hdr10', 'hlg', 'hdr10+', 'dolbyvision']
    """
    if G.KODI_VERSION < 20:  # The infolabel 'System.SupportedHDRTypes' is supported from Kodi v20
        return []
    # The infolabel System.SupportedHDRTypes returns the HDR types supported by the hardware as a string:
    # "HDR10, HLG, HDR10+, Dolby Vision"
    return xbmc.getInfoLabel('System.SupportedHDRTypes').replace(' ', '').lower().split(',')


def get_android_system_props():
    """Get Android system properties by parsing the raw output of getprop into a dictionary"""
    try:
        import subprocess
        info_dict = {}
        info = subprocess.check_output(['/system/bin/getprop']).decode('utf-8', errors='ignore').replace('\r\n', '\n')
        for line in info.split(']\n'):
            if not line:
                continue
            try:
                name, value = line.split(': ', 1)
            except ValueError:
                LOG.debug('Failed to parse getprop line: {}', line)
                continue
            name = name.strip()[1:-1]  # Remove brackets [] and spaces
            if value and value[0] == '[':
                value = value[1:]
            info_dict[name] = value
        return info_dict
    except OSError:
        LOG.error('Cannot get "getprop" data due to system error.')
        return {}