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}")