sex

app/services/external_api/tcgplayer/base_tcgplayer_service.py

@@ -0,0 +1,95 @@
+from typing import Any, Dict, Optional
+import aiohttp
+import logging
+from app.services.external_api.base_external_service import BaseExternalService
+from app.services.external_api.tcgplayer.tcgplayer_credentials import TCGPlayerCredentials
+
+logger = logging.getLogger(__name__)
+
+class BaseTCGPlayerService(BaseExternalService):
+    def __init__(self):
+        super().__init__(
+            store_base_url="https://store.tcgplayer.com",
+            login_endpoint="/oauth/login",
+            pricing_endpoint="/Admin/Pricing",
+            staged_inventory_endpoint=self.pricing_endpoint + "/DownloadStagedInventoryExportCSV?type=Pricing",
+            live_inventory_endpoint=self.pricing_endpoint + "/DownloadMyExportCSV?type=Pricing"
+        )
+        self.credentials = TCGPlayerCredentials()
+
+    def _get_headers(self, method: str) -> Dict[str, str]:
+        """Get headers based on the HTTP method"""
+        base_headers = {
+            'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
+            'accept-language': 'en-US,en;q=0.8',
+            'priority': 'u=0, i',
+            'referer': 'https://store.tcgplayer.com/admin/pricing',
+            'sec-ch-ua': '"Not A(Brand";v="8", "Chromium";v="132", "Brave";v="132"',
+            'sec-ch-ua-mobile': '?0', 
+            'sec-ch-ua-platform': '"macOS"',
+            'sec-fetch-dest': 'document',
+            'sec-fetch-mode': 'navigate',
+            'sec-fetch-site': 'same-origin',
+            'sec-fetch-user': '?1',
+            'sec-gpc': '1',
+            'upgrade-insecure-requests': '1',
+            'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36'
+        }
+        
+        if method == 'POST':
+            post_headers = {
+                'cache-control': 'max-age=0',
+                'content-type': 'application/x-www-form-urlencoded',
+                'origin': 'https://store.tcgplayer.com'
+            }
+            base_headers.update(post_headers)
+            
+        return base_headers
+
+    async def _make_request(
+        self,
+        method: str,
+        endpoint: str,
+        params: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
+        data: Optional[Dict[str, Any]] = None,
+        auth_required: bool = True
+    ) -> Dict[str, Any]:
+        session = await self._get_session()
+        url = f"{self.store_base_url}{endpoint}"
+        
+        # Get the authentication cookie if required
+        if auth_required:
+            cookie = self.credentials.get_cookie()
+            if not cookie:
+                raise RuntimeError("TCGPlayer authentication cookie not set. Please set the cookie using TCGPlayerCredentials.set_cookie()")
+            
+            # Get method-specific headers and update with any provided headers
+            request_headers = self._get_headers(method)
+            if headers:
+                request_headers.update(headers)
+            request_headers["Cookie"] = cookie
+        else:
+            request_headers = headers or {}
+        
+        try:
+            async with session.request(method, url, params=params, headers=request_headers, json=data) as response:
+                if response.status == 401:
+                    raise RuntimeError("TCGPlayer authentication failed. Cookie may be invalid or expired.")
+                response.raise_for_status()
+                return await response.json()
+        except aiohttp.ClientError as e:
+            logger.error(f"TCGPlayer API request failed: {str(e)}")
+            raise
+        except Exception as e:
+            logger.error(f"Unexpected error during TCGPlayer API request: {str(e)}")
+            raise
+
+    async def _get_session(self) -> aiohttp.ClientSession:
+        if self.session is None or self.session.closed:
+            self.session = aiohttp.ClientSession()
+        return self.session
+
+    async def close(self):
+        if self.session and not self.session.closed:
+            await self.session.close() 

app/services/external_api/tcgplayer/tcgplayer_credentials.py

@@ -0,0 +1,61 @@
+import os
+import json
+from typing import Optional
+import logging
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+class TCGPlayerCredentials:
+    _instance = None
+    _credentials_file = Path.home() / ".tcgplayer" / "credentials.json"
+    
+    def __new__(cls):
+        if cls._instance is None:
+            cls._instance = super(TCGPlayerCredentials, cls).__new__(cls)
+            cls._instance._initialize()
+        return cls._instance
+    
+    def _initialize(self):
+        """Initialize the credentials manager"""
+        self._cookie = None
+        self._load_credentials()
+    
+    def _load_credentials(self):
+        """Load credentials from the credentials file"""
+        try:
+            if self._credentials_file.exists():
+                with open(self._credentials_file, 'r') as f:
+                    data = json.load(f)
+                    self._cookie = data.get('cookie')
+        except Exception as e:
+            logger.error(f"Error loading TCGPlayer credentials: {str(e)}")
+    
+    def _save_credentials(self):
+        """Save credentials to the credentials file"""
+        try:
+            # Create directory if it doesn't exist
+            self._credentials_file.parent.mkdir(parents=True, exist_ok=True)
+            
+            with open(self._credentials_file, 'w') as f:
+                json.dump({'cookie': self._cookie}, f)
+            
+            # Set appropriate file permissions
+            self._credentials_file.chmod(0o600)
+        except Exception as e:
+            logger.error(f"Error saving TCGPlayer credentials: {str(e)}")
+    
+    def set_cookie(self, cookie: str):
+        """Set the authentication cookie"""
+        self._cookie = cookie
+        self._save_credentials()
+    
+    def get_cookie(self) -> Optional[str]:
+        """Get the authentication cookie"""
+        return self._cookie
+    
+    def clear_credentials(self):
+        """Clear stored credentials"""
+        self._cookie = None
+        if self._credentials_file.exists():
+            self._credentials_file.unlink() 

app/services/external_api/tcgplayer/tcgplayer_inventory_service.py

@@ -0,0 +1,18 @@
+from typing import Dict, List, Optional
+from app.services.external_api.tcgplayer.base_tcgplayer_service import BaseTCGPlayerService
+
+class TCGPlayerInventoryService(BaseTCGPlayerService):
+    def __init__(self):
+        super().__init__()
+
+    async def get_inventory(self) -> List[Dict]:
+        """Get inventory items"""
+        endpoint = "/inventory"
+        response = await self._make_request("GET", endpoint)
+        return response.get("results", [])
+
+    async def update_inventory(self, updates: List[Dict]) -> Dict:
+        """Update inventory items"""
+        endpoint = "/inventory"
+        response = await self._make_request("PUT", endpoint, data=updates)
+        return response

app/services/external_api/tcgplayer/tcgplayer_order_service.py

@@ -0,0 +1,106 @@
+from typing import Dict, List, Optional
+from datetime import datetime
+from app.services.external_api.tcgplayer.base_tcgplayer_service import BaseTCGPlayerService
+
+class TCGPlayerOrderService(BaseTCGPlayerService):
+    def __init__(self):
+        super().__init__()
+
+    async def get_orders(
+        self,
+        status: Optional[str] = None,
+        start_date: Optional[datetime] = None,
+        end_date: Optional[datetime] = None,
+        limit: int = 100
+    ) -> List[Dict]:
+        """
+        Get a list of orders with optional filtering
+        
+        Args:
+            status: Filter by order status (e.g., "Shipped", "Processing")
+            start_date: Filter orders after this date
+            end_date: Filter orders before this date
+            limit: Maximum number of orders to return
+            
+        Returns:
+            List of orders matching the criteria
+        """
+        endpoint = "/orders"
+        params = {"limit": limit}
+        
+        if status:
+            params["status"] = status
+        if start_date:
+            params["startDate"] = start_date.isoformat()
+        if end_date:
+            params["endDate"] = end_date.isoformat()
+            
+        response = await self._make_request("GET", endpoint, params=params)
+        return response.get("results", [])
+
+    async def get_order_details(self, order_id: str) -> Dict:
+        """
+        Get detailed information about a specific order
+        
+        Args:
+            order_id: TCGPlayer order ID
+            
+        Returns:
+            Detailed order information
+        """
+        endpoint = f"/orders/{order_id}"
+        response = await self._make_request("GET", endpoint)
+        return response
+
+    async def get_order_items(self, order_id: str) -> List[Dict]:
+        """
+        Get items in a specific order
+        
+        Args:
+            order_id: TCGPlayer order ID
+            
+        Returns:
+            List of items in the order
+        """
+        endpoint = f"/orders/{order_id}/items"
+        response = await self._make_request("GET", endpoint)
+        return response.get("results", [])
+
+    async def get_order_status(self, order_id: str) -> Dict:
+        """
+        Get the current status of an order
+        
+        Args:
+            order_id: TCGPlayer order ID
+            
+        Returns:
+            Order status information
+        """
+        endpoint = f"/orders/{order_id}/status"
+        response = await self._make_request("GET", endpoint)
+        return response
+
+    async def update_order_status(
+        self,
+        order_id: str,
+        status: str,
+        tracking_number: Optional[str] = None
+    ) -> Dict:
+        """
+        Update the status of an order
+        
+        Args:
+            order_id: TCGPlayer order ID
+            status: New status for the order
+            tracking_number: Optional tracking number for shipping
+            
+        Returns:
+            Updated order information
+        """
+        endpoint = f"/orders/{order_id}/status"
+        data = {"status": status}
+        if tracking_number:
+            data["trackingNumber"] = tracking_number
+            
+        response = await self._make_request("PUT", endpoint, data=data)
+        return response