pounce/backend/app/services/sedo_api.py

"""
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)
            
            # Check for error response
            if root.tag == "fault" or root.find(".//faultcode") is not None:
                fault_code = root.findtext(".//faultcode") or root.findtext("faultcode")
                fault_string = root.findtext(".//faultstring") or root.findtext("faultstring")
                return {"error": True, "faultcode": fault_code, "faultstring": fault_string}
            
            # Parse SEDOSEARCH response (domain listings)
            if root.tag == "SEDOSEARCH":
                items = []
                for item in root.findall("item"):
                    domain_data = {}
                    for child in item:
                        # Get the text content, handle type attribute
                        value = child.text
                        type_attr = child.get("type", "")
                        
                        # Convert types
                        if "double" in type_attr or "int" in type_attr:
                            try:
                                value = float(value) if value else 0
                            except:
                                pass
                        
                        domain_data[child.tag] = value
                    items.append(domain_data)
                
                return {"items": items, "count": len(items)}
            
            # Generic XML to dict fallback
            return self._xml_to_dict(root)
            
        except Exception as e:
            logger.warning(f"Failed to parse XML: {e}")
            return {"raw": xml_text, "error": str(e)}
    
    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 (XML parsed to dict).
        """
        params = {}
        
        if keyword:
            params["keyword"] = keyword
        if tld:
            params["tld"] = tld.lstrip(".")
        if min_price is not None:
            params["minprice"] = int(min_price)
        if max_price is not None:
            params["maxprice"] = int(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.
        
        Note: Sedo API doesn't have a dedicated auction filter.
        We filter by type='A' (auction) in post-processing.
        """
        params = {}
        
        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)
        
        result = await self._request("DomainSearch", params)
        
        # Filter to only show auctions (type='A')
        if "items" in result:
            result["items"] = [
                item for item in result["items"] 
                if item.get("type") == "A"
            ]
            result["count"] = len(result["items"])
        
        return result
    
    async def get_listings_for_display(
        self,
        keyword: Optional[str] = None,
        tld: Optional[str] = None,
        page_size: int = 50,
    ) -> List[Dict[str, Any]]:
        """
        Get Sedo listings formatted for display in Pounce.
        
        Returns a list of domains with affiliate URLs.
        """
        result = await self.search_domains(
            keyword=keyword,
            tld=tld,
            page_size=page_size
        )
        
        if "error" in result or "items" not in result:
            logger.warning(f"Sedo API error: {result}")
            return []
        
        listings = []
        for item in result.get("items", []):
            domain = item.get("domain", "")
            if not domain:
                continue
            
            # Get price (Sedo returns 0 for "Make Offer")
            price = item.get("price", 0)
            if isinstance(price, str):
                try:
                    price = float(price)
                except:
                    price = 0
            
            # Use the URL from Sedo (includes partner ID and tracking)
            url = item.get("url", f"https://sedo.com/search/details/?domain={domain}&partnerid={self.partner_id}")
            
            # Determine listing type
            listing_type = item.get("type", "D")  # D=Direct, A=Auction
            is_auction = listing_type == "A"
            
            listings.append({
                "domain": domain,
                "tld": domain.rsplit(".", 1)[1] if "." in domain else "",
                "price": price,
                "price_type": "bid" if is_auction else ("make_offer" if price == 0 else "fixed"),
                "is_auction": is_auction,
                "platform": "Sedo",
                "url": url,
                "rank": item.get("rank", 0),
            })
        
        return listings
    
    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()