""" Sedo Official API Client This service provides access to Sedo's official API for: - Domain search and auctions - Marketplace listings - Domain pricing API Documentation: https://api.sedo.com/apidocs/v1/ Type: XML-RPC based API SECURITY: - Credentials are loaded from environment variables - NEVER hardcode credentials in this file WHERE TO FIND YOUR CREDENTIALS: 1. Login to https://sedo.com 2. Go to "Mein Sedo" / "My Sedo" 3. Navigate to "API-Zugang" / "API Access" 4. You'll find: - Partner ID (your user ID) - SignKey (signature key for authentication) Usage: from app.services.sedo_api import sedo_client # Search domains for sale listings = await sedo_client.search_domains(keyword="tech") """ import logging import hashlib import time from datetime import datetime, timedelta from typing import Optional, List, Dict, Any import httpx from xml.etree import ElementTree from app.config import get_settings logger = logging.getLogger(__name__) class SedoAPIClient: """ Official Sedo API Client. Sedo uses an XML-RPC style API with signature-based authentication. Each request must include: - partnerid: Your partner ID - signkey: Your signature key (or hashed signature) """ def __init__(self): self.settings = get_settings() self.base_url = self.settings.sedo_api_base or "https://api.sedo.com/api/v1/" self.partner_id = self.settings.sedo_partner_id self.sign_key = self.settings.sedo_sign_key # HTTP client self._client: Optional[httpx.AsyncClient] = None @property def is_configured(self) -> bool: """Check if API credentials are configured.""" return bool(self.partner_id and self.sign_key) 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/x-www-form-urlencoded", "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 def _generate_signature(self, params: Dict[str, Any]) -> str: """ Generate request signature for Sedo API. The signature is typically: MD5(signkey + sorted_params) Check Sedo docs for exact implementation. """ # Simple implementation - may need adjustment based on actual Sedo requirements sorted_params = "&".join(f"{k}={v}" for k, v in sorted(params.items())) signature_base = f"{self.sign_key}{sorted_params}" return hashlib.md5(signature_base.encode()).hexdigest() async def _request( self, endpoint: str, params: Optional[Dict] = None ) -> Dict[str, Any]: """Make an authenticated API request.""" if not self.is_configured: raise ValueError("Sedo API credentials not configured") client = await self._get_client() # Base params for all requests request_params = { "partnerid": self.partner_id, "signkey": self.sign_key, **(params or {}) } url = f"{self.base_url.rstrip('/')}/{endpoint.lstrip('/')}" try: response = await client.get(url, params=request_params) response.raise_for_status() # Sedo API can return XML or JSON depending on endpoint content_type = response.headers.get("content-type", "") if "xml" in content_type: return self._parse_xml_response(response.text) elif "json" in content_type: return response.json() else: # Try JSON first, fallback to XML try: return response.json() except: return self._parse_xml_response(response.text) except httpx.HTTPError as e: logger.error(f"Sedo API request failed: {e}") raise def _parse_xml_response(self, xml_text: str) -> Dict[str, Any]: """Parse XML response from Sedo API.""" try: root = ElementTree.fromstring(xml_text) return self._xml_to_dict(root) except Exception as e: logger.warning(f"Failed to parse XML: {e}") return {"raw": xml_text} def _xml_to_dict(self, element) -> Dict[str, Any]: """Convert XML element to dictionary.""" result = {} for child in element: if len(child) > 0: result[child.tag] = self._xml_to_dict(child) else: result[child.tag] = child.text return result # ========================================================================= # DOMAIN SEARCH ENDPOINTS # ========================================================================= async def search_domains( self, keyword: Optional[str] = None, tld: Optional[str] = None, min_price: Optional[float] = None, max_price: Optional[float] = None, page: int = 1, page_size: int = 100, ) -> Dict[str, Any]: """ Search for domains listed on Sedo marketplace. Returns domains for sale (not auctions). """ params = { "output_method": "json", # Request JSON response } if keyword: params["keyword"] = 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 page: params["page"] = page if page_size: params["pagesize"] = min(page_size, 100) return await self._request("DomainSearch", params) async def search_auctions( self, keyword: Optional[str] = None, tld: Optional[str] = None, ending_within_hours: Optional[int] = None, page: int = 1, page_size: int = 100, ) -> Dict[str, Any]: """ Search for active domain auctions on Sedo. """ params = { "output_method": "json", "auction": "true", # Only auctions } if keyword: params["keyword"] = keyword if tld: params["tld"] = tld.lstrip(".") if page: params["page"] = page if page_size: params["pagesize"] = min(page_size, 100) return await self._request("DomainSearch", params) async def get_domain_details(self, domain: str) -> Dict[str, Any]: """Get detailed information about a specific domain.""" params = { "domain": domain, "output_method": "json", } return await self._request("DomainDetails", params) async def get_ending_soon_auctions( 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 ) # ========================================================================= # 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, "hint": "Find your credentials at: Sedo.com → Mein Sedo → API-Zugang" } try: # Try a simple search to test connection result = await self.search_domains(keyword="test", page_size=1) return { "success": True, "configured": True, "partner_id": self.partner_id, "authenticated_at": datetime.utcnow().isoformat() } except Exception as e: return { "success": False, "error": str(e), "configured": True } def transform_to_pounce_format(self, sedo_listing: Dict) -> Dict[str, Any]: """ Transform Sedo listing to Pounce internal format. Maps Sedo fields to our DomainAuction model. """ domain = sedo_listing.get("domain") or sedo_listing.get("domainname", "") tld = domain.rsplit(".", 1)[1] if "." in domain else "" # Parse end time if auction end_time_str = sedo_listing.get("auctionend") or sedo_listing.get("enddate") if end_time_str: try: end_time = datetime.fromisoformat(end_time_str.replace("Z", "+00:00")) except: end_time = datetime.utcnow() + timedelta(days=7) else: end_time = datetime.utcnow() + timedelta(days=7) # Price handling price = sedo_listing.get("price") or sedo_listing.get("currentbid") or 0 if isinstance(price, str): price = float(price.replace(",", "").replace("$", "").replace("€", "")) return { "domain": domain, "tld": tld, "platform": "Sedo", "current_bid": price, "buy_now_price": sedo_listing.get("buynow") or sedo_listing.get("bin"), "currency": sedo_listing.get("currency", "EUR"), "num_bids": sedo_listing.get("numbids") or sedo_listing.get("bidcount", 0), "end_time": end_time, "auction_url": f"https://sedo.com/search/details/?domain={domain}", "age_years": None, "reserve_met": sedo_listing.get("reservemet"), "traffic": sedo_listing.get("traffic"), "is_auction": sedo_listing.get("isaution") == "1" or sedo_listing.get("auction") == True, } # Singleton instance sedo_client = SedoAPIClient()