Replace urllib with requests

CHANGELOG.md

@@ -1,5 +1,10 @@
 # 2.x
 
+## [2.2.1](https://github.com/f-lawe/plugin.video.orange.fr/releases/tag/v2.2.1) - 2024-10-07
+
+### Changed
+- HTTP requests are now handled with [requests](https://pypi.org/project/requests/) ([#61](https://github.com/f-lawe/plugin.video.orange.fr/issues/61))
+
 ## [2.2.0](https://github.com/f-lawe/plugin.video.orange.fr/releases/tag/v2.2.0) - 2024-10-03
 
 ### Added

addon.xml

@@ -1,7 +1,8 @@
 <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
-<addon id="plugin.video.orange.fr" name="Orange TV France" version="2.2.0" provider-name="Flawe">
+<addon id="plugin.video.orange.fr" name="Orange TV France" version="2.2.1" provider-name="Flawe">
     <requires>
         <import addon="xbmc.python" version="3.0.1"/>
+        <import addon="script.module.requests" version="2.31.0"/>
         <import addon="script.module.routing" version="0.2.3"/>
         <import addon="script.module.inputstreamhelper" version="0.6.1"/>
     </requires>

requirements.txt

@@ -1,3 +1,4 @@
 kodi-addon-checker==0.0.*
 kodistubs==21.*
-ruff==0.3.*
+requests==2.31.0
+ruff==0.6.*

resources/lib/providers/abstract_orange_provider.py

@@ -5,14 +5,15 @@
 import re
 from abc import ABC
 from datetime import date, datetime, timedelta
-from http.client import HTTPSConnection
 from urllib.parse import urlencode
 
 import xbmc
+from requests import Session
+from requests.exceptions import RequestException
 
 from lib.providers.abstract_provider import AbstractProvider
 from lib.utils.kodi import DRM, build_addon_url, get_addon_setting, get_drm, get_global_setting, log
-from lib.utils.request import get_cookies, request, request_json, request_text, to_cookie_string
+from lib.utils.request import request, request_json, to_cookie_string
 
 _PROGRAMS_ENDPOINT = "https://rp-ott-mediation-tv.woopic.com/api-gw/live/v3/applications/STB4PC/programs?period={period}&epgIds=all&mco={mco}"
 _CATCHUP_CHANNELS_ENDPOINT = "https://rp-ott-mediation-tv.woopic.com/api-gw/catchup/v4/applications/PC/channels"
@@ -247,63 +248,56 @@ def _retrieve_auth_data(self, auth_url: str, login: str = None, password: str =
         """Retreive auth data from Orange (tv token and terminal id, plus wassup cookie when using credentials)."""
         cookies = {}
 
-        if login is not None and password is not None:
-            conn = HTTPSConnection("login.orange.fr")
-            res = request(conn, "https://login.orange.fr")
+        if login is not None or password is not None:
+            s = Session()
 
-            if res is None:
+            try:
+                res = request("GET", "https://login.orange.fr", s=s)
+                cookies = res.cookies.get_dict()
+            except RequestException:
                 log("Error while authenticating (init)", xbmc.LOGWARNING)
-                conn.close()
                 return None, None, None
 
-            cookies = get_cookies(res)
-            res.read()
-
-            res = request(
-                conn,
-                "https://login.orange.fr/api/login",
-                "POST",
-                headers={
-                    "Content-Type": "application/json",
-                    "Cookie": to_cookie_string(cookies, ["xauth"]),
-                },
-                body=json.dumps({"login": login, "params": {}, "isSosh": False}),
-            )
-
-            if res is None or res.status != 200:
-                log("Error while authenticating (login)", xbmc.LOGERROR)
-                conn.close()
+            try:
+                res = request(
+                    "POST",
+                    "https://login.orange.fr/api/login",
+                    headers={
+                        "Content-Type": "application/json",
+                        "Cookie": to_cookie_string(cookies, ["xauth"]),
+                    },
+                    data=json.dumps({"login": login, "params": {}, "isSosh": False}),
+                    s=s,
+                )
+                cookies = res.cookies.get_dict()
+            except RequestException:
+                log("Error while authenticating (login)", xbmc.LOGWARNING)
                 return None, None, None
 
-            cookies = get_cookies(res)
-            res.read()
-
-            res = request(
-                conn,
-                "https://login.orange.fr/api/password",
-                "POST",
-                headers={
-                    "Content-Type": "application/json",
-                    "Cookie": to_cookie_string(cookies, ["xauth"]),
-                },
-                body=json.dumps({"password": password, "remember": True}),
-            )
-
-            if res is None or res.status != 200:
-                log("Error while authenticating (password)", xbmc.LOGERROR)
-                conn.close()
+            try:
+                res = request(
+                    "POST",
+                    "https://login.orange.fr/api/password",
+                    headers={
+                        "Content-Type": "application/json",
+                        "Cookie": to_cookie_string(cookies, ["xauth"]),
+                    },
+                    data=json.dumps({"password": password, "params": {}}),
+                    s=s,
+                )
+                cookies = res.cookies.get_dict()
+            except RequestException:
+                log("Error while authenticating (password)", xbmc.LOGWARNING)
                 return None, None, None
 
-            cookies = get_cookies(res)
-            res.read()
-            conn.close()
-
-        html = request_text(auth_url, headers={"Cookie": to_cookie_string(cookies, ["trust", "wassup"])})
-
-        if html is None:
+        try:
+            res = request("GET", auth_url, headers={"Cookie": to_cookie_string(cookies, ["trust", "wassup"])})
+        except RequestException:
             log("Authentication page load failed", xbmc.LOGERROR)
             return None, None, None
 
+        html = res.text
+
         try:
             tv_token = re.search('instanceInfo:{token:"([a-zA-Z0-9-_.]+)"', html).group(1)
             household_id = re.search('householdId:"([A-Z0-9]+)"', html).group(1)

resources/lib/utils/kodi.py

@@ -15,8 +15,10 @@
 class DRM(Enum):
     """List DRM providers."""
 
+    CLEAR_KEY = "org.w3.clearkey"
+    PLAY_READY = "com.microsoft.playready"
     WIDEVINE = "com.widevine.alpha"
-    PLAYREADY = "com.microsoft.playready"
+    WISEPLAY = "com.huawei.wiseplay"
 
 
 def build_addon_url(path: str = "") -> str:

resources/lib/utils/request.py

@@ -1,16 +1,12 @@
 """Request utils."""
 
-import gzip
-import json
-from http.client import HTTPConnection, HTTPResponse, HTTPSConnection
-from http.cookies import SimpleCookie
+from http.client import HTTPConnection, HTTPSConnection
 from random import randint
-from socket import gaierror
 from typing import Mapping, TypeVar, Union
-from urllib.error import URLError
-from urllib.parse import unquote_plus, urlparse
 
 import xbmc
+from requests import Response, Session
+from requests.exceptions import JSONDecodeError, RequestException
 
 # from socks import SOCKS5
 # from sockshandler import SocksiPyHandler
@@ -39,111 +35,50 @@
 C = TypeVar("C", HTTPConnection, HTTPSConnection)
 
 
-def get_cookies(response: HTTPResponse) -> dict:
-    """Get cookies from HTTP response."""
-    cookies = {}
-    for header in response.getheaders():
-        if header[0] == "Set-Cookie":
-            cookie = header[1].split(";")[0]
-            cookies[cookie.split("=")[0]] = cookie.split("=")[1]
-    return cookies
-
-
 def get_random_ua() -> str:
     """Get a randomised user agent."""
     return _USER_AGENTS[randint(0, len(_USER_AGENTS) - 1)]
 
 
-def parse_cookies(cookie_strings: list) -> dict:
-    """Parse cookie strings."""
-    cookies = {}
-
-    for cookie_string in cookie_strings:
-        simple_cookie = SimpleCookie(cookie_string)
-        for key, item in simple_cookie.items():
-            cookies[key] = unquote_plus(item.value)
-
-    return cookies
-
-
-def request(conn: C, url: str, method: str = "GET", headers: Mapping[str, str] = None, body=None) -> C:
-    """Send HTTP request."""
+def request(method: str, url: str, headers: Mapping[str, str] = None, data=None, s: Session = None) -> Response:
+    """Send HTTP request using requests."""
     if headers is None:
         headers = {}
 
     headers = {
         "Accept": "*/*",
-        "Accept-Encoding": "br, gzip, deflate",
+        "Accept-Encoding": "gzip, deflate",
         "Accept-Language": "*",
         "Sec-Fetch-Mode": "cors",
         "User-Agent": get_random_ua(),
         **headers,
     }
 
-    try:
-        log(f"Fetching {url}", xbmc.LOGDEBUG)
-        conn.request(method, url, headers=headers, body=body)
-    except gaierror as e:
-        log(e, xbmc.LOGERROR)
-        return None
-    except URLError as e:
-        log(f"{e.reason}", xbmc.LOGERROR)
-        return None
-
-    res = conn.getresponse()
+    s = s if s is not None else Session()
 
-    if res.status != 200:
-        log(f"Error while fetching: {res.status} {res.reason}", xbmc.LOGERROR)
-        log(res.read(), xbmc.LOGDEBUG)
-        return None
-
-    return res
+    log(f"Fetching {url}", xbmc.LOGDEBUG)
+    return s.request(method, url, headers=headers, data=data)
 
 
 def request_json(url: str, headers: Mapping[str, str] = None, default=None) -> Union[dict, list]:
     """Send HTTP request and load json response."""
-    url = urlparse(url)
-    conn = HTTPConnection(url.netloc) if url.scheme == "http" else HTTPSConnection(url.netloc)
-    res = request(conn, url.geturl(), headers=headers)
-
-    if res is None:
-        conn.close()
+    try:
+        res = request("GET", url, headers=headers)
+        res.raise_for_status()
+    except RequestException as e:
+        log(e, xbmc.LOGERROR)
         return default
 
-    content = res.read()
-    conn.close()
-
-    if res.headers.get("Content-Encoding") == "gzip":
-        content = gzip.decompress(content)
-
     try:
-        content = json.loads(content)
-    except json.decoder.JSONDecodeError:
-        log("Cannot load json content", xbmc.LOGWARNING)
+        content = res.json()
+    except JSONDecodeError:
+        log("Cannot load json content", xbmc.LOGERROR)
+        log(res.text, xbmc.LOGDEBUG)
         return default
 
     return content
 
 
-def request_text(url: str, headers: Mapping[str, str] = None) -> str:
-    """Send HTTP request and load text response."""
-    url = urlparse(url)
-    conn = HTTPConnection(url.netloc) if url.scheme == "http" else HTTPSConnection(url.netloc)
-    res = request(conn, url.geturl(), headers=headers)
-
-    if res is None:
-        conn.close()
-        return None
-
-    content = res.read()
-    conn.close()
-
-    if res.headers.get("Content-Encoding") == "gzip":
-        content = gzip.decompress(content)
-
-    return content.decode("utf-8")
-
-
 def to_cookie_string(cookies: dict, pick: list = None) -> str:
     """Convert cookies to cookie string."""
     if pick is None: