parentzone_downloader/src/auth_manager.py

#!/usr/bin/env python3
"""
Authentication Manager for ParentZone API

This module handles authentication against the ParentZone login API
and manages session tokens for API requests.
"""

import asyncio
import aiohttp
import logging
from typing import Optional, Dict, Any
from urllib.parse import urljoin


class AuthManager:
    def __init__(self, api_url: str = "https://api.parentzone.me"):
        """
        Initialize the authentication manager.

        Args:
            api_url: Base URL of the API
        """
        self.api_url = api_url.rstrip("/")
        self.login_url = urljoin(self.api_url, "/v1/auth/login")
        self.create_session_url = urljoin(self.api_url, "/v1/auth/create-session")
        self.session_token: Optional[str] = None
        self.api_key: Optional[str] = None
        self.user_id: Optional[str] = None
        self.user_name: Optional[str] = None
        self.provider_name: Optional[str] = None
        self.logger = logging.getLogger(__name__)

        # Standard headers for login requests
        self.headers = {
            "accept": "application/json, text/plain, */*",
            "accept-language": "en-GB,en-US;q=0.9,en;q=0.8,ro;q=0.7",
            "content-type": "application/json;charset=UTF-8",
            "origin": "https://www.parentzone.me",
            "priority": "u=1, i",
            "sec-ch-ua": '"Not;A=Brand";v="99", "Google Chrome";v="139", "Chromium";v="139"',
            "sec-ch-ua-mobile": "?0",
            "sec-ch-ua-platform": '"macOS"',
            "sec-fetch-dest": "empty",
            "sec-fetch-mode": "cors",
            "sec-fetch-site": "same-site",
            "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36",
        }

    async def login(self, email: str, password: str) -> bool:
        """
        Login to the ParentZone API using two-step authentication.
        Step 1: Login with email/password to get user accounts
        Step 2: Create session with first account ID and password to get API key

        Args:
            email: User email
            password: User password

        Returns:
            True if login successful, False otherwise
        """
        self.logger.info(f"Attempting login for {email}")

        # Step 1: Login to get user accounts
        login_data = {"email": email, "password": password}

        timeout = aiohttp.ClientTimeout(total=30)
        async with aiohttp.ClientSession(timeout=timeout) as session:
            try:
                async with session.post(
                    self.login_url, headers=self.headers, json=login_data
                ) as response:
                    self.logger.info(f"Login response status: {response.status}")

                    if response.status == 200:
                        data = await response.json()
                        self.logger.info("Login successful")
                        self.logger.debug(f"Response data type: {type(data)}")
                        self.logger.debug(f"Full response data: {data}")

                        # Handle list response with user accounts
                        if isinstance(data, list) and len(data) > 0:
                            # Use the first account
                            first_account = data[0]
                            self.user_id = first_account.get("id")
                            self.user_name = first_account.get("name")
                            self.provider_name = first_account.get("providerName")

                            self.logger.info(
                                f"Selected account: {self.user_name} at {self.provider_name} (ID: {self.user_id})"
                            )

                            # Step 2: Create session with the account ID
                            return await self._create_session(password)
                        else:
                            self.logger.error(
                                f"Unexpected login response format: {data}"
                            )
                            return False
                    else:
                        error_text = await response.text()
                        self.logger.error(
                            f"Login failed with status {response.status}: {error_text}"
                        )
                        return False

            except Exception as e:
                self.logger.error(f"Login request failed: {e}")
                return False

    async def _create_session(self, password: str) -> bool:
        """
        Create a session using the user ID from login.

        Args:
            password: User password

        Returns:
            True if session creation successful, False otherwise
        """
        if not self.user_id:
            self.logger.error("No user ID available for session creation")
            return False

        self.logger.info(f"Creating session for user ID: {self.user_id}")

        session_data = {"id": self.user_id, "password": password}

        # Add x-api-product header for session creation
        session_headers = self.headers.copy()
        session_headers["x-api-product"] = "iConnect"

        timeout = aiohttp.ClientTimeout(total=30)
        async with aiohttp.ClientSession(timeout=timeout) as session:
            try:
                async with session.post(
                    self.create_session_url, headers=session_headers, json=session_data
                ) as response:
                    self.logger.info(
                        f"Create session response status: {response.status}"
                    )

                    if response.status == 200:
                        data = await response.json()
                        self.logger.info("Session creation successful")
                        self.logger.debug(f"Session response data: {data}")

                        # Extract API key from response
                        if isinstance(data, dict) and "key" in data:
                            self.api_key = data["key"]
                            self.logger.info("API key obtained successfully")
                            return True
                        else:
                            self.logger.error(
                                f"No 'key' field in session response: {data}"
                            )
                            return False
                    else:
                        error_text = await response.text()
                        self.logger.error(
                            f"Session creation failed with status {response.status}: {error_text}"
                        )
                        return False

            except Exception as e:
                self.logger.error(f"Session creation request failed: {e}")
                return False

    def get_auth_headers(self) -> Dict[str, str]:
        """
        Get headers with authentication token.

        Returns:
            Dictionary of headers including authentication
        """
        headers = self.headers.copy()

        if self.api_key:
            # Use x-api-key header for authenticated requests
            headers["x-api-key"] = self.api_key
            headers["x-api-product"] = "iConnect"

        return headers

    def is_authenticated(self) -> bool:
        """
        Check if currently authenticated.

        Returns:
            True if authenticated, False otherwise
        """
        return self.api_key is not None

    def logout(self):
        """Clear the session data."""
        self.api_key = None
        self.session_token = None
        self.user_id = None
        self.user_name = None
        self.provider_name = None
        self.logger.info("Logged out - session data cleared")


async def test_login():
    """Test the login functionality."""
    auth_manager = AuthManager()

    # Test credentials (replace with actual credentials)
    email = "tudor.sitaru@gmail.com"
    password = "mTVq8uNUvY7R39EPGVAm@"

    print("Testing ParentZone Login...")
    success = await auth_manager.login(email, password)

    if success:
        print("✅ Login successful!")
        print(f"User: {auth_manager.user_name} at {auth_manager.provider_name}")
        print(f"User ID: {auth_manager.user_id}")
        print(
            f"API Key: {auth_manager.api_key[:20]}..."
            if auth_manager.api_key
            else "No API key found"
        )

        # Test getting auth headers
        headers = auth_manager.get_auth_headers()
        print(f"Auth headers: {list(headers.keys())}")
    else:
        print("❌ Login failed!")


if __name__ == "__main__":
    asyncio.run(test_login())