OqeeRewind/utils/oqee.py

"""OQEE streaming service client for authentication and content access."""

import base64
from urllib.parse import urlparse, parse_qs
from bs4 import BeautifulSoup
import requests

from dotenv import load_dotenv
from utils.logging_config import logger

load_dotenv()


class OqeeClient:  # pylint: disable=too-many-instance-attributes
    """
    Service code for OQEE streaming service (https://oqee.com).

    Authorization: Credentials/IP
    Security: 1080p@L1 720@L3
    """

    def __init__(self, username: str, password: str):
        super().__init__()
        self.session = requests.Session()

        # Base headers template for API requests
        self._headers_template = {
            "accept": (
                "text/html,application/xhtml+xml,application/xml;q=0.9,"
                "image/avif,image/webp,image/apng,*/*;q=0.8,"
                "application/signed-exchange;v=b3;q=0.7"
            ),
            "accept-language": "en-GB,en-US;q=0.9,en;q=0.8",
            "cache-control": "no-cache",
            "upgrade-insecure-requests": "1",
            "user-agent": (
                "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
                "AppleWebKit/537.36 (KHTML, like Gecko) "
                "Chrome/144.0.0.0 Safari/537.36"
            ),
        }
        self.headers_base = self._build_headers()

        # Headers for API requests
        self.headers = None

        # Headers for manifest/licence
        self.headers_auth = None

        self.access_token = None
        self.right_token = None
        self.profil_id = None
        self.lic_url = "https://license.oqee.net/api/v1/live/license/widevine"
        if "fbx" in username.lower():
            self.abo = True
            self.fbx = True
        elif len(username) == 8 and username.isdigit():
            self.abo = True
            self.fbx = False
        else:
            self.abo = False

        self.configure(username, password)

    def certificate(self, **_):
        """
        Get the Service Privacy Certificate.
        """
        response = self.session.post(
            url=self.lic_url, headers=self.headers_auth, json={"licenseRequest": "CAQ="}
        )
        return response.json()["result"]["license"]

    def license(self, challenge, **_):
        """
        Get the License response for the specified challenge and title data.
        """
        license_request = base64.b64encode(challenge).decode()
        response = self.session.post(
            url=self.lic_url,
            headers=self.headers_auth,
            json={"licenseRequest": license_request},
        )
        if not response.json()["success"]:
            raise ValueError(
                f"License request failed: {response.json()['error']['msg']}"
            )
        return response.json()["result"]["license"]

    def configure(self, username, password):
        """Configure the client by logging in and processing title information."""
        logger.info("Logging in")
        self.login(username, password)

    def _build_headers(self, overrides=None, remove=None):
        """Clone default headers and apply optional overrides/removals."""
        headers = self._headers_template.copy()
        if overrides:
            headers.update(overrides)
        if remove:
            for key in remove:
                headers.pop(key, None)
        return headers

    def right(self):
        """
        Get user rights token from Oqee API.
        """
        headers = self._build_headers(
            overrides={"authorization": f"Bearer {self.access_token}"}
        )
        data = self.session.get(
            "https://api.oqee.net/api/v3/user/rights_proxad", headers=headers
        ).json()
        return data["result"]["token"]

    def profil(self):
        """
        Gets the first profile ID from the OQEE API.
        """
        headers = self._build_headers(
            overrides={"authorization": f"Bearer {self.access_token}"}
        )
        data = self.session.get(
            "https://api.oqee.net/api/v2/user/profiles", headers=headers
        ).json()
        # logger.info("Selecting first profile by default.")
        return data["result"][0]["id"]

    def login_cred_fbx(self, username, password):
        """Authenticate with OQEE service using Free account credentials."""
        headers = self._build_headers(
            overrides={
                "accept-language": "fr-FR,fr;q=0.8",
                "cache-control": "no-cache",
                "pragma": "no-cache",
                "priority": "u=1, i",
                "sec-ch-ua": '"Brave";v="131", "Chromium";v="131", "Not_A Brand";v="24"',
                "user-agent": (
                    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
                    "AppleWebKit/537.36 (KHTML, like Gecko) "
                    "Chrome/131.0.0.0 Safari/537.36"
                ),
                "x-oqee-customization": "0",
            }
        )
        data = {"provider": "free", "platform": "web"}
        response = self.session.post(
            "https://api.oqee.net/api/v2/user/oauth/init", headers=headers, json=data
        ).json()
        redirect_url = response["result"]["redirect_url"]
        r = parse_qs(urlparse(redirect_url).query)
        client_id = r["client_id"][0]
        redirect_uri = r["redirect_uri"][0]
        state = r["state"][0]

        headers = {
            "Accept": (
                "text/html,application/xhtml+xml,application/xml;q=0.9, "
                "image/avif,image/webp,image/apng,*/*;q=0.8"
            ),
            "Accept-Language": "fr-FR,fr;q=0.7",
            "Cache-Control": "max-age=0",
            "Connection": "keep-alive",
            "Content-Type": "application/x-www-form-urlencoded",
            "Origin": "https://subscribe.free.fr",
            "Referer": "https://subscribe.free.fr/auth/auth.pl?",
            "Upgrade-Insecure-Requests": "1",
            "user-agent": (
                "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
                "AppleWebKit/537.36 (KHTML, like Gecko) "
                "Chrome/131.0.0.0 Safari/537.36"
            ),
        }
        data = {
            "login": username,
            "pass": password,
            "ok": "Se connecter",
            "client_id": client_id,
            "ressource": "",
            "response_type": "code",
            "redirect_uri": redirect_uri,
            "state": state,
        }
        r = self.session.post(
            "https://subscribe.free.fr/auth/auth.pl", headers=headers, data=data
        )
        parsed_url = parse_qs(urlparse(r.url).query)
        if "result" not in parsed_url:
            raise ValueError(
                "Login failed: invalid credentials or error in authentication"
            )
        token = parsed_url["result"][0]

        headers = self._build_headers(
            overrides={"x-oqee-customization": "0"}, remove=("x-oqee-account-provider",)
        )
        data = self.session.post(
            "https://api.oqee.net/api/v5/user/login",
            headers=headers,
            json={"type": "freeoa", "token": token},
        ).json()
        return data["result"]["token"]

    def login_cred_mobile(self, username, password):
        """Authenticate with OQEE service using Free mobile account credentials."""
        headers = {
            "accept": "*/*",
            "accept-language": "en-GB,en-US;q=0.9,en;q=0.8",
            "cache-control": "no-cache",
            "content-type": "application/json",
            "origin": "https://tv.free.fr",
            "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36",
            "x-oqee-customization": "0",
            "x-oqee-platform": "web",
        }

        data = {"provider": "freemobile", "platform": "web"}
        response = self.session.post(
            "https://api.oqee.net/api/v2/user/oauth/init", headers=headers, json=data
        ).json()
        redirect_url = response["result"]["redirect_url"]
        token = parse_qs(urlparse(redirect_url).query)["token"][0]

        headers = {
            "Content-Type": "application/x-www-form-urlencoded",
            "Origin": "https://mobile.free.fr",
            "Upgrade-Insecure-Requests": "1",
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36",
        }

        page = self.session.get(redirect_url, headers=headers)
        soup = BeautifulSoup(page.text, "html.parser")

        token_csrf = soup.find("input", {"name": "token-csrf"})
        token_csrf_value = token_csrf["value"] if token_csrf else None
        callback_url = soup.find("input", {"name": "callback_url"})
        callback_url_value = callback_url["value"] if callback_url else None

        params = (
            ("service", "oqee"),
            ("token", token),
        )

        data = {
            "token-csrf": token_csrf_value,
            "callback_url": callback_url_value,
            "identifiant": username,
            "password": password,
            "action": "sso-submit",
        }

        response = self.session.post(
            "https://mobile.free.fr/sso/", headers=headers, params=params, data=data
        )

        parsed_url = parse_qs(urlparse(response.url).query)
        if "result" not in parsed_url:
            raise ValueError(
                "Login failed: invalid credentials or error in authentication"
            )
        freemobile_token = parsed_url["result"][0]

        headers = self._build_headers(
            overrides={"x-oqee-customization": "0"}, remove=("x-oqee-account-provider",)
        )
        data = self.session.post(
            "https://api.oqee.net/api/v5/user/login",
            headers=headers,
            json={"type": "freemobileoa", "token": freemobile_token},
        ).json()
        return data["result"]["token"]

    def login_cred_free(self, username, password):
        """Authenticate with OQEE service using Free account credentials."""
        headers = {
            "accept": "*/*",
            "accept-language": "en-GB,en-US;q=0.9,en;q=0.8",
            "cache-control": "no-cache",
            "content-type": "application/json",
            "origin": "https://tv.free.fr",
            "user-agent": (
                "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
                "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36"
            ),
            "x-oqee-customization": "0",
            "x-oqee-platform": "web",
        }

        data = {"provider": "oqee", "platform": "web"}
        response = self.session.post(
            "https://api.oqee.net/api/v2/user/oauth/init", headers=headers, json=data
        ).json()
        redirect_url = response["result"]["redirect_url"]
        token = parse_qs(urlparse(redirect_url).query)["token"][0]

        data = {"email": username, "password": password, "token": token}
        response = self.session.post(
            "https://api.oqee.net/api/v1/user/oauthorize",
            headers=headers,
            json=data,
        ).json()

        if not response["success"]:
            error_msg = response.get("error", {}).get("msg", "No error message")
            raise ValueError(
                f"Login failed: invalid credentials or error in authentication - {error_msg}"
            )
        parsed_url = parse_qs(urlparse(response["result"]["redirect_url"]).query)
        if "code" not in parsed_url:
            raise ValueError(
                "Login failed: invalid credentials or error in authentication "
                "- no code in redirect URL"
            )
        code = parsed_url["code"][0]

        data = self.session.post(
            "https://api.oqee.net/api/v5/user/login",
            headers=headers,
            json={"type": "oqeeoa", "token": code},
        ).json()
        return data["result"]["token"]

    def login_ip(self):
        """
        Performs IP-based authentication with the OQEE service.
        """
        headers = self._build_headers(
            overrides={"x-oqee-customization": "0"}, remove=("x-oqee-account-provider",)
        )
        data = {"type": "ip"}
        data = self.session.post(
            "https://api.oqee.net/api/v5/user/login", headers=headers, json=data
        ).json()
        return data["result"]["token"]

    def login(self, username, password):
        """
        Log in to the Oqee service and set up necessary tokens and headers.
        """
        if not username or not password:
            logger.info("No credentials provided, using IP login.")
            self.access_token = self.login_ip()
        else:
            try:
                if self.abo:
                    if self.fbx:
                        logger.info(
                            "Logging in with provided credentials (abo account detected)."
                        )
                        self.access_token = self.login_cred_fbx(username, password)
                    else:
                        logger.info(
                            "Logging in with provided credentials (abo account detected, non-FBX)."
                        )
                        self.access_token = self.login_cred_mobile(username, password)
                else:
                    logger.info(
                        "Logging in with provided credentials (free account detected)."
                    )
                    self.access_token = self.login_cred_free(username, password)
            except ValueError as e:
                logger.warning(
                    "Credential login failed: %s. Falling back to IP login.", e
                )
                self.access_token = self.login_ip()

        logger.info("Fetching rights token")
        self.right_token = self.right()
        logger.debug(
            "Rights token obtained: %s",
            self.right_token[:10] + "..." if self.right_token else "None",
        )
        logger.info("Fetching profile ID")
        self.profil_id = self.profil()
        logger.debug("Profile ID obtained: %s", self.profil_id)

        self.headers = self._build_headers(
            overrides={
                "x-fbx-rights-token": self.right_token,
                "x-oqee-profile": self.profil_id,
            }
        )

        self.headers_auth = self._build_headers(
            overrides={
                "x-fbx-rights-token": self.right_token,
                "x-oqee-profile": self.profil_id,
                "authorization": f"Bearer {self.access_token}",
            }
        )