[plugin.video.orange.fr] v2.3.5 (#4656)

Author: f-lawe

plugin.video.orange.fr/CHANGELOG.md

@@ -1,5 +1,15 @@
 # 2.x
 
+## [2.3.5](https://github.com/f-lawe/plugin.video.orange.fr/releases/tag/v2.3.5) - 2025-04-18
+
+### Fixed
+- Auth and live stream update following new Orange TV website changes
+
+## [2.3.4](https://github.com/f-lawe/plugin.video.orange.fr/releases/tag/v2.3.4) - 2025-04-08
+
+### Changed
+- New logo in order to avoid confusion with the new Orange Radio
+
 ## [2.3.3](https://github.com/f-lawe/plugin.video.orange.fr/releases/tag/v2.3.3) - 2025-01-15
 
 ### Changed

plugin.video.orange.fr/README.md

@@ -10,7 +10,7 @@ _Cet addon n'est pas officiellement supporté par Orange. Tous les produits, log
 ## Description
 Cet addon ajoute la TV d'Orange à Kodi. Toutes les chaînes inclues dans votre abonnement sont maintenant directement accessibles depuis Kodi !
 
-Cet addon inclue :
+Cet addon inclut :
 - la télé en direct dans Kodi TV (en utilisant [IPTV Simple PVR](https://github.com/kodi-pvr/pvr.iptvsimple))
 - les programmes en replay
 - l'accès aux chaînes payantes qui font partie de votre souscription (pour le direct et le replay)

plugin.video.orange.fr/addon.xml

@@ -1,5 +1,5 @@
 <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
-<addon id="plugin.video.orange.fr" name="Orange TV France" version="2.3.3" provider-name="Flawe">
+<addon id="plugin.video.orange.fr" name="Orange TV France" version="2.3.5" provider-name="Flawe">
   <requires>
     <import addon="xbmc.python" version="3.0.1"/>
     <import addon="script.module.requests" version="2.31.0"/>
@@ -22,7 +22,7 @@
     <source>https://github.com/f-lawe/plugin.video.orange.fr</source>
     <email>francois@lavaud.family</email>
     <assets>
-      <icon>resources/media/icon.png</icon>
+      <icon>resources/media/orange_tv.png</icon>
       <fanart>resources/media/fanart.jpg</fanart>
       <screenshot>resources/media/screenshot_1.jpg</screenshot>
       <screenshot>resources/media/screenshot_2.jpg</screenshot>

plugin.video.orange.fr/resources/lib/providers/abstract_orange_provider.py

@@ -24,14 +24,15 @@
 _CATCHUP_VIDEOS_ENDPOINT = "https://rp-ott-mediation-tv.woopic.com/api-gw/catchup/v4/applications/PC/groups/{group_id}"
 _CHANNELS_ENDPOINT = "https://rp-ott-mediation-tv.woopic.com/api-gw/pds/v1/live/ew?everywherePopulation=OTT_Metro"
 
-_LIVE_STREAM_ENDPOINT = "https://mediation-tv.orange.fr/all/api-gw/live/v3/auth/accountToken/applications/PC/channels/{stream_id}/stream?terminalModel=WEB_PC&terminalId="
+_LIVE_STREAM_ENDPOINT = "https://mediation-tv.orange.fr/all/api-gw/stream/v2/auth/accountToken/live/{stream_id}?deviceModel=WEB_PC&customerOrangePopulation=OTT_Metro"
 _CATCHUP_STREAM_ENDPOINT = "https://mediation-tv.orange.fr/all/api-gw/catchup/v4/auth/accountToken/applications/PC/videos/{stream_id}/stream?terminalModel=WEB_PC&terminalId="
 
 _STREAM_LOGO_URL = "https://proxymedia.woopic.com/api/v1/images/2090{path}"
-_LIVE_HOMEPAGE_URL = "https://chaines-tv.orange.fr/"
-_CATCHUP_VIDEO_URL = "https://replay.orange.fr/videos/{stream_id}"
+_LIVE_HOMEPAGE_URL = "https://tv.orange.fr/"
 _LOGIN_URL = "https://login.orange.fr"
 
+_LICENSE_ENDPOINT = "https://mediation-tv.orange.fr/all/api-gw/license/v1/auth/accountToken"
+
 
 class AbstractOrangeProvider(AbstractProvider, ABC):
     """Abstract Orange Provider."""
@@ -42,16 +43,15 @@ class AbstractOrangeProvider(AbstractProvider, ABC):
 
     def get_live_stream_info(self, stream_id: str) -> dict:
         """Get live stream info."""
-        auth_url = _LIVE_HOMEPAGE_URL
-        return self._get_stream_info(auth_url, _LIVE_STREAM_ENDPOINT, stream_id)
+        return self._get_stream_info(_LIVE_STREAM_ENDPOINT, stream_id)
 
     def get_catchup_stream_info(self, stream_id: str) -> dict:
         """Get catchup stream info."""
-        auth_url = _CATCHUP_VIDEO_URL.format(stream_id=stream_id)
-        return self._get_stream_info(auth_url, _CATCHUP_STREAM_ENDPOINT, stream_id)
+        return self._get_stream_info(_CATCHUP_STREAM_ENDPOINT, stream_id)
 
     def get_streams(self) -> list:
         """Load stream data from Orange and convert it to JSON-STREAMS format."""
+        # @todo: use new API to check if channel is part of subscription
         channels = request_json(_CHANNELS_ENDPOINT, default={"channels": {}})["channels"]
         channels.sort(key=lambda channel: channel["displayOrder"])
 
@@ -63,12 +63,16 @@ def get_streams(self) -> list:
                 "name": channel["name"],
                 "preset": str(channel["displayOrder"]),
                 "logo": self._extract_logo(channel["logos"]),
-                "stream": build_addon_url(f"/stream/live/{channel['idEPG']}"),
+                "stream": build_addon_url(f"/stream/live/{self._get_channel_live_id(channel)}"),
                 "group": [group_name for group_name in self.groups if int(channel["idEPG"]) in self.groups[group_name]],
             }
             for channel in channels
         ]
 
+    def _get_channel_live_id(self, channel: dict) -> str:
+        """Get live id for given channel."""
+        return channel["technicalChannels"]["live"][0]["liveTargetURLRelativePath"]
+
     def get_epg(self) -> dict:
         """Load EPG data from Orange and convert it to JSON-EPG format."""
         past_days_to_display = get_global_setting("epg.pastdaystodisplay", int)
@@ -198,9 +202,9 @@ def _get_catchup_videos(self, channel_id: str, category_id: str, article_id: str
             for video in videos
         ]
 
-    def _get_stream_info(self, auth_url: str, stream_endpoint: str, stream_id: str) -> dict:
+    def _get_stream_info(self, stream_endpoint: str, stream_id: str) -> dict:
         """Load stream info from Orange."""
-        tv_token, tv_token_expires, wassup = self._retrieve_auth_data(auth_url)
+        tv_token, tv_token_expires, wassup = self._retrieve_auth_data()
 
         try:
             stream_endpoint_url = stream_endpoint.format(stream_id=stream_id)
@@ -221,14 +225,16 @@ def _get_stream_info(self, auth_url: str, stream_endpoint: str, stream_id: str)
     def _compute_stream_info(self, stream: dict, tv_token: str, wassup: str) -> dict:
         """Compute stream info."""
         protectionData = stream.get("protectionData") or stream.get("protectionDatas")
-        license_server_url = None
+        path = stream.get("streamURL") or stream.get("url")
+
+        license_server_url = _LICENSE_ENDPOINT if stream.get("url") is None else ""
 
         for system in protectionData:
             if system.get("keySystem") == get_drm():
-                license_server_url = system.get("laUrl")
+                license_server_url += system.get("laUrl")
 
         stream_info = {
-            "path": stream.get("url"),
+            "path": path,
             "protocol": "mpd",
             "mime_type": "application/xml+dash",
             "drm_config": {
@@ -253,7 +259,7 @@ def _compute_stream_info(self, stream: dict, tv_token: str, wassup: str) -> dict
         log(stream_info, xbmc.LOGDEBUG)
         return stream_info
 
-    def _retrieve_auth_data(self, auth_url: str, login: str = None, password: str = None) -> (str, str, str):
+    def _retrieve_auth_data(self, login: str = None, password: str = None) -> (str, str, str):
         """Retreive auth data from Orange (tv token and wassup cookie)."""
         provider_session_data = get_addon_setting("provider.session_data", dict)
         tv_token, tv_token_expires, wassup = (
@@ -268,13 +274,14 @@ def _retrieve_auth_data(self, auth_url: str, login: str = None, password: str =
                 session.headers["Cookie"] = f"wassup={wassup}"
 
             try:
-                response = request("GET", f"{_LIVE_HOMEPAGE_URL}token", s=session)
-            except RequestException:
+                response = request("GET", _LIVE_HOMEPAGE_URL, session=session)
+                tv_token = json.loads(re.search('"token":(".*?")', response.text).group(1))
+            except AttributeError:
                 log("Login required", xbmc.LOGINFO)
                 self._login(session)
-                response = request("GET", f"{_LIVE_HOMEPAGE_URL}token", s=session)
+                response = request("GET", _LIVE_HOMEPAGE_URL, session=session)
+                tv_token = json.loads(re.search('"token":(".*?")', response.text).group(1))
 
-            tv_token = response.json()
             tv_token_expires = datetime.utcnow().timestamp() + 30 * 60
 
             if "wassup" in session.cookies:
@@ -309,7 +316,7 @@ def _login(self, session):
         }
 
         try:
-            request("GET", _LOGIN_URL, headers=session.headers, s=session)
+            request("GET", _LOGIN_URL, headers=session.headers, session=session)
         except RequestException:
             log("Error while authenticating (init)", xbmc.LOGWARNING)
             return
@@ -319,7 +326,7 @@ def _login(self, session):
                 "POST",
                 f"{_LOGIN_URL}/api/login",
                 data=json.dumps({"login": login, "params": {}}),
-                s=session,
+                session=session,
             )
         except RequestException:
             log("Error while authenticating (login)", xbmc.LOGWARNING)

plugin.video.orange.fr/resources/lib/utils/request.py

@@ -36,7 +36,7 @@ def get_random_ua() -> str:
     return _USER_AGENTS[randint(0, len(_USER_AGENTS) - 1)]
 
 
-def request(method: str, url: str, headers: Mapping[str, str] = None, data=None, s: Session = None) -> Response:
+def request(method: str, url: str, headers: Mapping[str, str] = None, data=None, session: Session = None) -> Response:
     """Send HTTP request using requests."""
     if headers is None:
         headers = {}
@@ -50,10 +50,10 @@ def request(method: str, url: str, headers: Mapping[str, str] = None, data=None,
         **headers,
     }
 
-    s = s if s is not None else Session()
+    session = session if session is not None else Session()
 
     log(f"Fetching {url}", xbmc.LOGDEBUG)
-    res = s.request(method, url, headers=headers, data=data)
+    res = session.request(method, url, headers=headers, data=data)
     res.raise_for_status()
     log(f" -> {res.status_code}", xbmc.LOGDEBUG)
     return res