""" DropCatch Official API Client This service provides access to DropCatch's official API for: - Searching domain auctions - Getting auction details - Backorder management API Documentation: https://www.dropcatch.com/hiw/dropcatch-api Interactive Docs: https://api.dropcatch.com/swagger SECURITY: - Credentials are loaded from environment variables - NEVER hardcode credentials in this file Usage: from app.services.dropcatch_api import dropcatch_client # Get active auctions auctions = await dropcatch_client.search_auctions(keyword="tech") """ import logging from datetime import datetime, timedelta from typing import Optional, List, Dict, Any import httpx from functools import lru_cache from app.config import get_settings logger = logging.getLogger(__name__) class DropCatchAPIClient: """ Official DropCatch API Client. This uses the V2 API endpoints (V1 is deprecated). Authentication is via OAuth2 client credentials. """ def __init__(self): self.settings = get_settings() self.base_url = self.settings.dropcatch_api_base or "https://api.dropcatch.com" self.client_id = self.settings.dropcatch_client_id self.client_secret = self.settings.dropcatch_client_secret # Token cache self._access_token: Optional[str] = None self._token_expires_at: Optional[datetime] = None # HTTP client self._client: Optional[httpx.AsyncClient] = None @property def is_configured(self) -> bool: """Check if API credentials are configured.""" return bool(self.client_id and self.client_secret) async def _get_client(self) -> httpx.AsyncClient: """Get or create HTTP client.""" if self._client is None or self._client.is_closed: self._client = httpx.AsyncClient( timeout=30.0, headers={ "Content-Type": "application/json", "User-Agent": "Pounce/1.0 (Domain Intelligence Platform)" } ) return self._client async def close(self): """Close the HTTP client.""" if self._client and not self._client.is_closed: await self._client.aclose() self._client = None async def _authenticate(self) -> str: """ Authenticate with DropCatch API and get access token. POST https://api.dropcatch.com/authorize Body: { "clientId": "...", "clientSecret": "..." } Returns: Access token string """ if not self.is_configured: raise ValueError("DropCatch API credentials not configured") # Check if we have a valid cached token if self._access_token and self._token_expires_at: if datetime.utcnow() < self._token_expires_at - timedelta(minutes=5): return self._access_token client = await self._get_client() try: response = await client.post( f"{self.base_url}/authorize", json={ "clientId": self.client_id, "clientSecret": self.client_secret } ) if response.status_code != 200: logger.error(f"DropCatch auth failed: {response.status_code} - {response.text}") raise Exception(f"Authentication failed: {response.status_code}") data = response.json() # Extract token - the response format may vary # Common formats: { "token": "...", "expiresIn": 3600 } # or: { "accessToken": "...", "expiresIn": 3600 } self._access_token = data.get("token") or data.get("accessToken") or data.get("access_token") # Calculate expiry (default 1 hour if not specified) expires_in = data.get("expiresIn") or data.get("expires_in") or 3600 self._token_expires_at = datetime.utcnow() + timedelta(seconds=expires_in) logger.info("DropCatch API: Successfully authenticated") return self._access_token except httpx.HTTPError as e: logger.error(f"DropCatch auth HTTP error: {e}") raise async def _request( self, method: str, endpoint: str, params: Optional[Dict] = None, json_data: Optional[Dict] = None ) -> Dict[str, Any]: """Make an authenticated API request.""" token = await self._authenticate() client = await self._get_client() headers = { "Authorization": f"Bearer {token}" } url = f"{self.base_url}{endpoint}" try: response = await client.request( method=method, url=url, params=params, json=json_data, headers=headers ) if response.status_code == 401: # Token expired, re-authenticate self._access_token = None token = await self._authenticate() headers["Authorization"] = f"Bearer {token}" response = await client.request( method=method, url=url, params=params, json=json_data, headers=headers ) response.raise_for_status() return response.json() except httpx.HTTPError as e: logger.error(f"DropCatch API request failed: {e}") raise # ========================================================================= # AUCTION ENDPOINTS (V2) # ========================================================================= async def search_auctions( self, keyword: Optional[str] = None, tld: Optional[str] = None, min_price: Optional[float] = None, max_price: Optional[float] = None, ending_within_hours: Optional[int] = None, page_size: int = 100, page_token: Optional[str] = None, ) -> Dict[str, Any]: """ Search for domain auctions. Endpoint: GET /v2/auctions (or similar - check interactive docs) Returns: { "auctions": [...], "cursor": { "next": "...", "previous": "..." } } """ params = { "pageSize": page_size, } if keyword: params["searchTerm"] = keyword if tld: params["tld"] = tld.lstrip(".") if min_price is not None: params["minPrice"] = min_price if max_price is not None: params["maxPrice"] = max_price if ending_within_hours: params["endingWithinHours"] = ending_within_hours if page_token: params["pageToken"] = page_token return await self._request("GET", "/v2/auctions", params=params) async def get_auction(self, auction_id: int) -> Dict[str, Any]: """Get details for a specific auction.""" return await self._request("GET", f"/v2/auctions/{auction_id}") async def get_ending_soon( self, hours: int = 24, page_size: int = 50 ) -> Dict[str, Any]: """Get auctions ending soon.""" return await self.search_auctions( ending_within_hours=hours, page_size=page_size ) async def get_hot_auctions(self, page_size: int = 50) -> Dict[str, Any]: """ Get hot/popular auctions (high bid activity). Note: The actual endpoint may vary - check interactive docs. """ # This might be a different endpoint or sort parameter params = { "pageSize": page_size, "sortBy": "bidCount", # or "popularity" - check docs "sortOrder": "desc" } return await self._request("GET", "/v2/auctions", params=params) # ========================================================================= # BACKORDER ENDPOINTS (V2) # ========================================================================= async def search_backorders( self, keyword: Optional[str] = None, page_size: int = 100, page_token: Optional[str] = None, ) -> Dict[str, Any]: """Search for available backorders (domains dropping soon).""" params = {"pageSize": page_size} if keyword: params["searchTerm"] = keyword if page_token: params["pageToken"] = page_token return await self._request("GET", "/v2/backorders", params=params) # ========================================================================= # UTILITY METHODS # ========================================================================= async def test_connection(self) -> Dict[str, Any]: """Test the API connection and credentials.""" if not self.is_configured: return { "success": False, "error": "API credentials not configured", "configured": False } try: await self._authenticate() return { "success": True, "configured": True, "client_id": self.client_id.split(":")[0] if ":" in self.client_id else self.client_id, "authenticated_at": datetime.utcnow().isoformat() } except Exception as e: return { "success": False, "error": str(e), "configured": True } def transform_to_pounce_format(self, dc_auction: Dict) -> Dict[str, Any]: """ Transform DropCatch auction to Pounce internal format. Maps DropCatch fields to our DomainAuction model. """ domain = dc_auction.get("domainName") or dc_auction.get("domain", "") tld = domain.rsplit(".", 1)[1] if "." in domain else "" # Parse end time (format may vary) end_time_str = dc_auction.get("auctionEndTime") or dc_auction.get("endTime") if end_time_str: try: end_time = datetime.fromisoformat(end_time_str.replace("Z", "+00:00")) except: end_time = datetime.utcnow() + timedelta(days=1) else: end_time = datetime.utcnow() + timedelta(days=1) return { "domain": domain, "tld": tld, "platform": "DropCatch", "current_bid": dc_auction.get("currentBid") or dc_auction.get("price", 0), "currency": "USD", "num_bids": dc_auction.get("bidCount") or dc_auction.get("numberOfBids", 0), "end_time": end_time, "auction_url": f"https://www.dropcatch.com/domain/{domain}", "age_years": dc_auction.get("yearsOld") or dc_auction.get("age"), "buy_now_price": dc_auction.get("buyNowPrice"), "reserve_met": dc_auction.get("reserveMet"), "traffic": dc_auction.get("traffic"), "external_id": str(dc_auction.get("auctionId") or dc_auction.get("id", "")), } # Singleton instance dropcatch_client = DropCatchAPIClient()