giga_tcg/app/services/util/_requests.py

from typing import Dict, Optional
from app.services.util._docker import DockerUtil
from enum import Enum
import browser_cookie3
import os
import json
import requests
import time
from datetime import datetime
import logging

logger = logging.getLogger(__name__)

class Browser(Enum):
    """Supported browser types for cookie extraction"""
    BRAVE = "brave"
    CHROME = "chrome"
    FIREFOX = "firefox"

class Method(Enum):
    """Supported HTTP methods"""
    GET = "GET"
    POST = "POST"

class TCGPlayerEndpoints(Enum):
    """Supported TCGPlayer API endpoints"""
    ORDERS = "https://order-management-api.tcgplayer.com/orders"

class Headers:
    ACCEPT = 'application/json, text/plain, */*'
    ACCEPT_ENCODING = 'gzip, deflate, br, zstd'
    ACCEPT_LANGUAGE = 'en-US,en;q=0.8'
    PRIORITY = 'u=1, i'
    SECCHUA = '"Not(A:Brand";v="99", "Brave";v="133", "Chromium";v="133"'
    SECCHUA_MOBILE = '?0'
    SECCHUA_PLATFORM = '"macOS"'
    SEC_FETCH_DEST = 'empty'
    SEC_FETCH_MODE = 'cors'
    SEC_FETCH_SITE = 'same-site'
    SEC_GPC = '1'
    USER_AGENT = 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36'
    SELLER_ORIGIN = 'https://sellerportal.tcgplayer.com'
    SELLER_REFERER = 'https://sellerportal.tcgplayer.com/'

class RequestHeaders:
    BASE_HEADERS = {
        'accept': Headers.ACCEPT,
        'accept-encoding': Headers.ACCEPT_ENCODING,
        'accept-language': Headers.ACCEPT_LANGUAGE,
        'priority': Headers.PRIORITY,
        'sec-ch-ua': Headers.SECCHUA,
        'sec-ch-ua-mobile': Headers.SECCHUA_MOBILE,
        'sec-ch-ua-platform': Headers.SECCHUA_PLATFORM,
        'sec-fetch-dest': Headers.SEC_FETCH_DEST,
        'sec-fetch-mode': Headers.SEC_FETCH_MODE,
        'sec-fetch-site': Headers.SEC_FETCH_SITE,
        'sec-gpc': Headers.SEC_GPC,
        'user-agent': Headers.USER_AGENT
    }
    SELLER_HEADERS = {
        'origin': Headers.SELLER_ORIGIN,
        'referer': Headers.SELLER_REFERER
    }
    POST_HEADERS = {
        'content-type': 'application/json'
    }

class URLHeaders:
    # combine base and seller headers
    ORDER_HEADERS = {**RequestHeaders.BASE_HEADERS, **RequestHeaders.SELLER_HEADERS}
    POST_HEADERS = {**RequestHeaders.BASE_HEADERS, **RequestHeaders.SELLER_HEADERS, **RequestHeaders.POST_HEADERS}

class RequestsUtil:
    def __init__(self, browser_type: Browser = Browser.BRAVE):
        self.browser_type = browser_type
        self.docker_util = DockerUtil()
        self.previous_request_time = datetime.now()
    
    def get_session(self, cookies: Dict = None) -> requests.Session:
        """Create a session with the specified cookies"""
        session = requests.Session()
        if cookies:
            session.cookies.update(cookies)
        return session
    
    def bare_request(self, url: str, method: str, cookies: dict = None, data=None) -> requests.Response:
        """Send a request without any additional processing"""
        try:
            response = requests.request(method, url, cookies=cookies, data=data)
            response.raise_for_status()
            return response
        except requests.RequestException as e:
            logger.error(f"Request failed: {str(e)}")
            return None

    def get_tcgplayer_cookies_from_file(self) -> Dict:
        # check if cookies file exists
        if not os.path.exists('cookies/tcg_cookies.json'):
            raise ValueError("Cookies file not found")
        with open('cookies/tcg_cookies.json', 'r') as f:
            logger.debug("Loading cookies from file")
            cookies = json.load(f)
        return cookies
    
    def get_tcgplayer_browser_cookies(self) -> Optional[Dict]:
        """Retrieve cookies from the specified browser"""
        try:
            cookie_getter = getattr(browser_cookie3, self.browser_type.value, None)
            if not cookie_getter:
                raise ValueError(f"Unsupported browser type: {self.browser_type.value}")
            return cookie_getter()
        except Exception as e:
            logger.error(f"Failed to get browser cookies: {str(e)}")
            return None
    
    def rate_limit(self, time_between_requests: int = 10):
        """Rate limit requests by waiting for a specified time between requests"""
        time_diff = (datetime.now() - self.previous_request_time).total_seconds()
        if time_diff < time_between_requests:
            time.sleep(time_between_requests - time_diff)
    
    def send_request(self, url: str, method: str, cookies: dict,  data=None, json=None) -> requests.Response:
        """Send a request with the specified cookies"""
                
        headers = self.set_headers(url, method)
        if not headers:
            raise ValueError("Headers not set")

        try:
            self.rate_limit()
            response = requests.request(method, url, headers=headers, cookies=cookies, data=data, json=json)
            response.raise_for_status()
            self.previous_request_time = datetime.now()
            
            return response

        except requests.RequestException as e:
            logger.error(f"Request failed: {str(e)}")
            return None
        
    def set_headers(self, url: str, method: str) -> Dict:
        # use tcgplayerendpoints enum to set headers where url partially matches enum value
        for endpoint in TCGPlayerEndpoints:
            if endpoint.value in url and str.upper(method) == "POST":
                return URLHeaders.POST_HEADERS
            elif endpoint.value in url:
                return URLHeaders.ORDER_HEADERS
            else:
                raise ValueError(f"Endpoint not found in TCGPlayerEndpoints: {url}")