magentaeu: calculate transaction id

lib/streaming_providers/providers/magentaeu/epg_manager.py

@@ -33,6 +33,7 @@ from datetime import datetime, timedelta, timezone
 from zoneinfo import ZoneInfo
 from typing import Any, Dict, List, Optional, Set, Tuple
 
+from .utils import build_guest_headers
 from ...base.utils.logger import logger
 from ...base.models.epg_models import EPGEntry, EPGProgramDetails
 from .constants import (
@@ -40,7 +41,6 @@ from .constants import (
     SUPPORTED_COUNTRIES,
     get_app_key,
     get_bifrost_url,
-    get_guest_headers,
     get_language,
     get_natco_key,
 )
@@ -375,22 +375,10 @@ class MagentaEUEpgManager:
         return device_id, session_id
 
     def _guest_headers(self, flow: str, step: str) -> Dict[str, str]:
-        """Build request headers for a guest (unauthenticated) bifrost call."""
         device_id, session_id = self._current_ids()
-
-        logger.info(f"[MagentaEUEpgManager] Using device_id: {device_id}")
-        logger.info(f"[MagentaEUEpgManager] Using session_id: {session_id}")
-
-        headers = get_guest_headers(self._country, device_id, session_id)
-        headers.update({
-            "x-call-time": str(int(time.time() * 1000)),  # Unix timestamp in ms
-            "x-tv-flow": flow,
-            "x-tv-step": step,
-            "x-txn-id": uuid.uuid4().hex,                 # Transaction ID
-            "x-request-tracking-id": str(uuid.uuid4()),
-        })
-        logger.debug(f"[MagentaEUEpgManager] Request headers: {headers}")
-        return headers
+        return build_guest_headers(
+            self._country, device_id, session_id, flow=flow, step=step
+        )
 
     @staticmethod
     def _ensure_tz(dt: datetime) -> datetime:

lib/streaming_providers/providers/magentaeu/provider.py

@@ -2,7 +2,6 @@
 # -*- coding: utf-8 -*-
 import time
 import datetime
-import uuid
 from typing import ClassVar, Dict, List, Optional, Tuple
 
 from ...base.auth import UserPasswordCredentials
@@ -36,7 +35,6 @@ from .constants import (
     WV_URL,
     get_base_url,
     get_bifrost_url,
-    get_guest_headers,
     get_language,
     get_natco_key,
 )
@@ -182,14 +180,11 @@ class MagentaEUProvider(StreamingProvider):
                 bifrost_url=get_bifrost_url(self.country)
             )
 
-            headers = get_guest_headers(self.country, device_id, session_id)
-            headers.update({
-                "x-call-time": str(int(time.time() * 1000)),
-                "x-tv-flow": "START_UP",
-                "x-tv-step": "EPG_CHANNEL",
-                "x-txn-id": uuid.uuid4().hex,
-                "x-request-tracking-id": str(uuid.uuid4()),
-            })
+            from .utils import  build_guest_headers
+
+            headers = build_guest_headers(
+                self.country, device_id, session_id, flow="START_UP"
+            )
 
             params = {
                 "channelMap_id": "",

lib/streaming_providers/providers/magentaeu/utils.py

@@ -0,0 +1,40 @@
+# streaming_providers/providers/magentaeu/utils.py
+
+import hashlib
+import time
+import uuid
+from typing import Dict, Optional
+from .constants import get_guest_headers
+
+
+def _generate_txn_id(
+    tracking_id: str, session_id: str, device_id: str, call_time: str
+) -> str:
+    """SHA-256(trackingId + sessionId + deviceId + callTime)[:32]"""
+    raw = tracking_id + session_id + device_id + call_time
+    return hashlib.sha256(raw.encode()).hexdigest()[:32]
+
+
+def build_guest_headers(
+    country: str,
+    device_id: str,
+    session_id: str,
+    flow: str,
+    step: Optional[str] = None,
+    tracking_id: Optional[str] = None,
+) -> Dict[str, str]:
+    if tracking_id is None:
+        tracking_id = str(uuid.uuid4())
+
+    # Snapshot call_time once — x-txn-id is derived from it, so both
+    # headers must use the same value.
+    call_time = str(int(time.time() * 1000))
+
+    headers = get_guest_headers(country, device_id, session_id)
+    headers["x-call-time"] = call_time
+    headers["x-tv-flow"] = flow
+    headers["x-request-tracking-id"] = tracking_id
+    headers["x-txn-id"] = _generate_txn_id(tracking_id, session_id, device_id, call_time)
+    if step is not None:
+        headers["x-tv-step"] = step
+    return headers